第4章 用户行为数据采集模块

4.1 数据通道

4.2 环境准备

4.2.1 集群所有进程查看脚本

1）在/home/atguigu/bin目录下创建脚本xcall

[atguigu@hadoop102 bin]$ vim xcall

2）在脚本中编写如下内容

#! /bin/bashfor i in hadoop102 hadoop103 hadoop104
doecho --------- $i ----------ssh $i "$*"
done

3）修改脚本执行权限

[atguigu@hadoop102 bin]$ chmod 777 xcall

4）启动脚本

[atguigu@hadoop102 bin]$ xcall jps

4.2.2 Hadoop安装

1）安装步骤

详见：大数据框架保姆级安装教程——hadoop

（1）集群规划

hadoop102 hadoop103 hadoop104
HDFS NameNode
DataNode 		DataNode DataNode
SecondaryNameNode
YARN NodeManager Resourcemanager
NodeManager 		NodeManager

注意：尽量使用离线方式安装

2）项目经验

（1）项目经验之HDFS存储多目录

①生产环境服务器磁盘情况

②在hdfs-site.xml文件中配置多目录，注意新挂载磁盘的访问权限问题

HDFS的DataNode节点保存数据的路径由dfs.datanode.data.dir参数决定，其默认值为file://${hadoop.tmp.dir}/dfs/data，若服务器有多个磁盘，必须对该参数进行修改。如服务器磁盘如上图所示，则该参数应修改为如下的值。

<property><name>dfs.datanode.data.dir</name><value>file:///dfs/data1,file:///hd2/dfs/data2,file:///hd3/dfs/data3,file:///hd4/dfs/data4</value>
</property>

注意：因为每台服务器节点的磁盘情况不同，所以这个配置配完之后，不需要分发

（2）项目经验之集群数据均衡

①节点间数据均衡

开启数据均衡命令：

start-balancer.sh -threshold 10

对于参数10，代表的是集群中各个节点的磁盘空间利用率相差不超过10%，可根据实际情况进行调整。

停止数据均衡命令

stop-balancer.sh

注意：于HDFS需要启动单独的Rebalance Server来执行Rebalance操作，所以尽量不要在NameNode上执行start-balancer.sh，而是找一台比较空闲的机器。

②磁盘间数据均衡

生成均衡计划（我们只有一块磁盘，不会生成计划）

hdfs diskbalancer -plan hadoop103

执行均衡计划

hdfs diskbalancer -execute hadoop103.plan.json

查看当前均衡任务的执行情况

hdfs diskbalancer -query hadoop103

取消均衡任务

hdfs diskbalancer -cancel hadoop103.plan.json

（3）项目经验之Hadoop参数调优

①HDFS参数调优hdfs-site.xml

The number of Namenode RPC server threads that listen to requests from clients. If dfs.namenode.servicerpc-address is not configured then Namenode RPC server threads listen to requests from all nodes.
NameNode有一个工作线程池，用来处理不同DataNode的并发心跳以及客户端并发的元数据操作。
对于大集群或者有大量客户端的集群来说，通常需要增大参数dfs.namenode.handler.count的默认值10。
<property><name>dfs.namenode.handler.count</name><value>10</value>
</property>

dfs.namenode.handler.count=20×〖log〗_e^(Cluster Size)，比如集群规模为8台时，此参数设置为41。可通过简单的python代码计算该值，代码如下。

[atguigu@hadoop102 ~]$ python
Python 2.7.5 (default, Apr 11 2018, 07:36:10)
[GCC 4.8.5 20150623 (Red Hat 4.8.5-28)] on linux2
Type "help", "copyright", "credits" or "license" for more information.
>>> import math
>>> print int(20*math.log(8))
41
>>> quit()

②YARN参数调优yarn-site.xml

情景描述：总共7台机器，每天几亿条数据，数据源->Flume->Kafka->HDFS->Hive

面临问题：数据统计主要用HiveSQL，没有数据倾斜，小文件已经做了合并处理，开启的JVM重用，而且IO没有阻塞，内存用了不到50%。但是还是跑的非常慢，而且数据量洪峰过来时，整个集群都会宕掉。基于这种情况有没有优化方案。

解决办法：

内存利用率不够。这个一般是Yarn的2个配置造成的，单个任务可以申请的最大内存大小，和Hadoop单个节点可用内存大小。调节这两个参数能提高系统内存的利用率。

（a）yarn.nodemanager.resource.memory-mb

表示该节点上YARN可使用的物理内存总量，默认是8192（MB），注意，如果你的节点内存资源不够8GB，则需要调减小这个值，而YARN不会智能的探测节点的物理内存总量。

（b）yarn.scheduler.maximum-allocation-mb

单个任务可申请的最多物理内存量，默认是8192（MB）。

4.2.3 Zookeeper安装

1）安装步骤

详见：大数据框架保姆级安装教程——Zookeeper

4.2.4 Kafka安装

1）安装步骤

详见：大数据框架保姆级安装教程——Kafka（3.0.0）

4.2.5 Flume安装

按照采集通道规划，需在hadoop102，hadoop103，hadoop104三台节点分别部署一个Flume。可参照以下步骤先在hadoop102安装，然后再进行分发。

1）安装步骤

详见：大数据框架保姆级安装教程——Flume

2）分发Flume

[atguigu@hadoop102 ~]$ xsync /opt/module/flume/

3）项目经验

（1）堆内存调整

Flume堆内存通常设置为4G或更高，配置方式如下：

修改/opt/module/flume/conf/flume-env.sh文件，配置如下参数**（虚拟机环境暂不配置）**

export JAVA_OPTS="-Xms4096m -Xmx4096m -Dcom.sun.management.jmxremote"

注：

-Xms表示JVM Heap(堆内存)最小尺寸，初始分配；

-Xmx 表示JVM Heap(堆内存)最大允许的尺寸，按需分配。

4.3 日志采集Flume

4.3.1 日志采集Flume配置概述

按照规划，需要采集的用户行为日志文件分布在hadoop102，hadoop103两台日志服务器，故需要在hadoop102，hadoop103两台节点配置日志采集Flume。日志采集Flume需要采集日志文件内容，并对日志格式（JSON）进行校验，然后将校验通过的日志发送到Kafka。

此处可选择TaildirSource和KafkaChannel，并配置日志校验拦截器。

选择TailDirSource和KafkaChannel的原因如下：

1）TailDirSource

TailDirSource相比ExecSource、SpoolingDirectorySource的优势

TailDirSource：断点续传、多目录。Flume1.6以前需要自己自定义Source记录每次读取文件位置，实现断点续传。

ExecSource可以实时搜集数据，但是在Flume不运行或者Shell命令出错的情况下，数据将会丢失。

SpoolingDirectorySource监控目录，支持断点续传。

2）KafkaChannel

采用Kafka Channel，省去了Sink，提高了效率。

日志采集Flume关键配置如下：

4.3.2 日志采集Flume配置实操

1）创建Flume配置文件

在hadoop102节点的Flume的job目录下创建file_to_kafka.conf

[atguigu@hadoop102 flume]$ mkdir job
[atguigu@hadoop102 flume]$ vim job/file_to_kafka.conf

2）配置文件内容如下

#定义组件
a1.sources = r1
a1.channels = c1#配置source
a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /opt/module/applog/log/app.*
a1.sources.r1.positionFile = /opt/module/flume/taildir_position.json
a1.sources.r1.interceptors =  i1
a1.sources.r1.interceptors.i1.type = com.atguigu.gmall.flume.interceptor.ETLInterceptor$Builder#配置channel
a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092
a1.channels.c1.kafka.topic = topic_log
a1.channels.c1.parseAsFlumeEvent = false#组装
a1.sources.r1.channels = c1

3）编写Flume拦截器

（1）创建Maven工程flume-interceptor

（2）创建包：com.atguigu.gmall.flume.interceptor

（3）在pom.xml文件中添加如下配置

<dependencies><dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-core</artifactId><version>1.9.0</version><scope>provided</scope></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.62</version></dependency>
</dependencies><build><plugins><plugin><artifactId>maven-compiler-plugin</artifactId><version>2.3.2</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin><plugin><artifactId>maven-assembly-plugin</artifactId><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins>
</build>

（4）在com.atguigu.gmall.flume.utils包下创建JSONUtil类

package com.atguigu.gmall.flume.utils;import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONException;public class JSONUtil {/*
* 通过异常判断是否是json字符串
* 是：返回true  不是：返回false
* */public static boolean isJSONValidate(String log){try {JSONObject.parseObject(log);return true;}catch (JSONException e){return false;}}
}

（5）在com.atguigu.gmall.flume.interceptor包下创建ETLInterceptor类

package com.atguigu.gmall.flume.interceptor;import com.atguigu.gmall.flume.utils.JSONUtil;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.List;public class ETLInterceptor implements Interceptor {@Overridepublic void initialize() {}@Overridepublic Event intercept(Event event) {//1、获取body当中的数据并转成字符串byte[] body = event.getBody();String log = new String(body, StandardCharsets.UTF_8);//2、判断字符串是否是一个合法的json，是：返回当前event；不是：返回nullif (JSONUtil.isJSONValidate(log)) {return event;} else {return null;}}@Overridepublic List<Event> intercept(List<Event> list) {Iterator<Event> iterator = list.iterator();while (iterator.hasNext()){Event next = iterator.next();if(intercept(next)==null){iterator.remove();}}return list;}public static class Builder implements Interceptor.Builder{@Overridepublic Interceptor build() {return new ETLInterceptor();}@Overridepublic void configure(Context context) {}}@Overridepublic void close() {}
}

（6）打包

（7）需要先将打好的包放入到hadoop102的/opt/module/flume/lib文件夹下面。

4.3.3 日志采集Flume测试

1）启动Zookeeper、Kafka集群

2）启动hadoop102的日志采集Flume

[atguigu@hadoop102 flume]$ bin/flume-ng agent -n a1 -c conf/ -f job/file_to_kafka.conf -Dflume.root.logger=info,console

3）启动一个Kafka的Console-Consumer

[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic topic_log

4）生成模拟数据

[atguigu@hadoop102 ~]$ lg.sh

5）观察Kafka消费者是否能消费到数据

4.3.4 日志采集Flume启停脚本

1）分发日志采集Flume配置文件和拦截器

若上述测试通过，需将hadoop102节点的Flume的配置文件和拦截器jar包，向另一台日志服务器发送一份。

[atguigu@hadoop102 flume]$ scp -r job hadoop103:/opt/module/flume/
[atguigu@hadoop102 flume]$ scp lib/flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar hadoop103:/opt/module/flume/lib/

2）方便起见，此处编写一个日志采集Flume进程的启停脚本

在hadoop102节点的/home/atguigu/bin目录下创建脚本f1.sh

[atguigu@hadoop102 bin]$ vim f1.sh

在脚本中填写如下内容

#!/bin/bashcase $1 in
"start"){for i in hadoop102 hadoop103doecho " --------启动 $i 采集flume-------"ssh $i "nohup /opt/module/flume/bin/flume-ng agent -n a1 -c /opt/module/flume/conf/ -f /opt/module/flume/job/file_to_kafka.conf >/dev/null 2>&1 &"done
};;
"stop"){for i in hadoop102 hadoop103doecho " --------停止 $i 采集flume-------"ssh $i "ps -ef | grep file_to_kafka | grep -v grep |awk  '{print \$2}' | xargs -n1 kill -9 "done};;
esac

3）增加脚本执行权限

[atguigu@hadoop102 bin]$ chmod 777 f1.sh

4）f1启动

[atguigu@hadoop102 module]$ f1.sh start

5）f2停止

[atguigu@hadoop102 module]$ f1.sh stop

大数据项目实战之数据仓库：用户行为采集平台——第4章 用户行为数据采集模块相关推荐

  1. 大数据项目实战教程：使用SparkSQL+Hbase+Oozie构建企业级用户画像

    大数据项目实战教程,本课程需要有大数据基础(掌握基本大数据组件应用)的人才可以学习哦!市面上全面的大数据教程较少,今天分享给大家的就是一套全面的大数据学习教程,企业级大数据项目:360度用户画像实战 ...

  2. 大数据项目实战——基于某招聘网站进行数据采集及数据分析（四）

    大数据项目实战 第四章 数据预处理 文章目录 大数据项目实战 学习目标 一.分析预处理数据 1)salary 2)city 3)skillLabels 4)companyLabelList.posit ...

  3. 【PBL项目实战】户外智慧农场项目实战系列——7.Mind+Mixly双平台工业级多合一空气质量传感器数据上云及云端可视化展示

    [PBL项目实战]户外智慧农场项目实战系列--7.Mind+Mixly双平台工业级多合一空气质量传感器数据上云及云端可视化展示 原文链接  https://mp.weixin.qq.com/s/ZEU ...

  4. 大数据项目之_15_电信客服分析平台_0102_项目背景+项目架构+项目实现+数据生产+数据采集/消费(存储)

    大数据项目之_15_电信客服分析平台_01&02 一.项目背景 二.项目架构 三.项目实现 3.1.数据生产 3.1.1.数据结构 3.1.2.编写代码 3.1.3.打包测试 3.2.数据采集 ...

  5. 【PBL项目实战】户外智慧农场项目实战系列——1.阿里云物联网平台的开通与云端可视化应用的新建

    岭师人工智能素养教育共同体PBL项目教程系列之1:户外智慧农场项目实战系列 原文链接  https://mp.weixin.qq.com/s/avKWJHNvnPeFsdEXUBByjw 本系列以户外 ...

  6. 【课题总结】OpenCV 抠图项目实战（11）算法实验平台

    Python 小白的课题报告-OpenCV 抠图项目实战(11)抠图算法实验平台 本系列是 Python 小白的课题作业<基于OpenCV 的图像分割和抠图>. 需要说明的是,本系列并不能 ...

  7. 1. 用户行为采集平台概述

    1. 用户行为采集平台概述 数据仓库概念 项目需求及架构设计 项目需求分析 项目框架 技术选型 系统数据流程设计 框架版本选型 服务器选型 集群资源规划设计 用户行为日志 用户行为日志概述 用户行为日 ...

  8. 1、电商数仓（用户行为采集平台）

    数据仓库概念 数据仓库( Data Warehouse ),是为企业制定决策,提供数据支持的.可以帮助企业,改进业务流程.提高产品质量等. 数据仓库的输入数据通常包括:业务数据.用户行为数据和爬虫数据 ...

  9. 大数据开发实战：数据仓库技术

    1.OLTP和OLAP OLTP的全称是 Online Transaction Processing, OLTP主要用传统的关系型数据库来进行事务处理.OLTP最核心的需求是单条记录的高效快速处理,索 ...

最新文章

  1. leetcode算法题--数组中数字出现的次数
  2. MSTAR GAMMA
  3. Origin教程：DSC曲线的描述和绘制
  4. Spark _04集群搭建及测试
  5. Servlet其他关联类---ServletConfig类
  6. power bi示例文件_Power BI桌面问答数据交互示例
  7. 【BZOJ3174】[TJOI2013]拯救小矮人（贪心，动态规划）
  8. 测试用例设计设计方法——正交实验法
  9. 阿里云短信接口方法使用
  10. 计算机音乐b型谱简单,尤克里里入门教程_尤克里里谱大全
  11. saas系统和传统服务器的区别,传统CRM与SaaS模式的区别
  12. JSHOP2学习4：浅谈与SHOP2的区别
  13. Excel常用技巧笔记
  14. 九宫格拼图android代码,九宫格拼图代码
  15. c语言怎么表示字节集,用易语言写一个电子钢琴软件
  16. 【无标题】发的东方人
  17. PCF8591详解(蓝桥杯单片机模块(IIC总线))
  18. latex 公式换行、对齐
  19. Linux建立软链接、硬链接
  20. 面试官扎心一问：NIO的原理是什么机制？

热门文章

  1. PRN（20210426）：GRAPH-BASED CONTINUAL LEARNING(ICLR2021)
  2. word文字怎么竖向排列_word文档如何将文字竖着排列？word文字排列技巧
  3. MongoDB数据库注入-墨者学院(SQL手工注入漏洞测试(MongoDB数据库))
  4. 美丽和创意的单页产品组合网站
  5. 新建数据中心的项目管理和经验分享
  6. miniui.open方法(通过子页面将数据带入到主页面内)
  7. Word处理控件Aspose.Words功能演示：在 Python 中将 Word 文档转换为 PNG、JPEG 或 BMP
  8. 【Bootstrap】常用组件（框架）
  9. 设置vs2019护眼编程背景色
  10. cc2530之外部中断实验