Flume 数据采集
模块
1.
1
数据处理链路
1.2
环境准备
1
.
2
.
1
集群
进程查看
脚本
（1）在/home/bigdata_admin/bin目录下创建脚本xcall.sh
[bigdata_admin@hadoop102
bin]$ vim xcall.sh
（2）在脚本中编写如下内容
#! /bin/bash
for i in hadoop102 hadoop103 hadoop104
do
echo --------- $i ----------
ssh $i "$*"
done（3）修改脚本执行权限
[bigdata_admin@hadoop102 bin
]$ chmod 777 xcall.sh
（4）启动脚本
[bigdata_admin@hadoop102 bin
]$ xcall.sh jps
1
.
2.2
H
adoop
安装
1）安装步骤
略
2）项目经验
（1）项目经验之HDFS存储多目录
虚拟机的项目不需要配置，我们只有一个磁盘。
1.生产环境服务器磁盘情况
2.在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）项目经验之集群数据均衡
1节点间数据均衡
开启数据均衡命令。
start-balancer.sh -threshold 10
对于参数10，代表的是集群中各个节点的磁盘空间利用率相差不超过10%，可根据实际情况进行调整。
停止数据均衡命令。
stop-balancer.sh
2磁盘间数据均衡
生成均衡计划
（我们只有一块磁盘，不会生成计划）。
hdfs diskbalancer -plan hadoop103
执行均衡计划。
hdfs diskbalancer -execute hadoop103.plan.json
查看当前均衡任务的执行情况。
hdfs diskbalancer -query hadoop103
取消均衡任务。
hdfs diskbalancer -cancel hadoop103.plan.json
（3）项目经验之Hadoop参数调优
1.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)(ClusterSize)，比如集群规模为8台时，此参数设置为41。可通过简单的python代码计算该值，代码如下。
[bigdata_admin@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()
2.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）。
1
.
2.3
Zookeeper安装
1
）安装步骤
略
2
）
ZK
集群启动停止脚本
（1）在hadoop102的/home/bigdata_admin/bin目录下创建脚本
[bigdata_admin@hadoop102 bin]$ vim zk.sh
在脚本中编写如下内容。
#!/bin/bash
case $1 in
"start"){
for i in hadoop102 hadoop103 hadoop104
do
echo ---------- zookeeper $i 启动 ------------
ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh start"
done
};;
"stop"){
for i in hadoop102 hadoop103 hadoop104
do
echo ---------- zookeeper $i 停止 ------------
ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh stop"
done
};;
"status"){
for i in hadoop102 hadoop103 hadoop104
do
echo ---------- zookeeper $i 状态 ------------
ssh $i "/opt/module/zookeeper-3.5.7/bin/zkServer.sh status"
done
};;
esac（2）增加脚本执行权限
[bigdata_admin@hadoop102 bin]$ chmod 777 zk.sh
（3）Zookeeper集群启动脚本
[bigdata_admin@hadoop102 module]$ zk.sh start
（4）Zookeeper集群停止脚本
[bigdata_admin@hadoop102 module]$ zk.sh stop
1
.
2.4
Kafk
a
安装
1
）安装步骤
略
1
.2
.
5 Flume
安装
按照采集通道规划，需在hadoop102，hadoop103，hadoop104三台节点分别部署一个Flume。
可参照以下步骤先在hadoop
102
安装，然后再进行分发。
1）安装步骤
略
2）分发Flume到hadoop
103
、hadoop
104
[bigdata_admin@hadoop102 ~]$ xsync /opt/module/flume/
3）项目经验
（1）堆内存调整
Flume堆内存通常设置为4G或更高，配置方式如下：
修改/opt/module/flume/conf/flume-env.sh文件，配置如下参数
（虚拟机环境暂不配置）
export JAVA_
OPTS
=”-Xms
4096
m -Xmx
4096
m -Dcom.sun.management.jmxremote”
注：
-Xms表示JVM Heap（堆内存）最小尺寸，初始分配；
-Xmx 表示JVM Heap（堆内存）最大允许的尺寸，按需分配。
1
.
3
日志采集Flume
1.
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）Kafka
Channel
采用Kafka
Channel，
省去了Sink
，提高了
效率。
日志采集Flume关键配置如下：
1
.3
.
2
日志采集Flume配置实操
1）创建Flume配置文件
（1）在hadoop102节点的Flume的job目录下创建file_to_kafka.conf。
[bigdata_admin@hadoop102 flume]$ mkdir job
[bigdata_admin@hadoop102 flume]$ vim job/file_to_kafka.conf
配置文件内容如下：
#为各组件命名
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.bigdata_admin.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
a1.channels.c1.kafka.topic = topic_log
a1.channels.c1.parseAsFlumeEvent = false
#绑定source和channel以及sink和channel的关系
a1.sources.r1.channels = c1（2）分发配置文件到hadoop103
[bigdata_admin@hadoop102 flume]$ xsync job
2
）编写Flume拦截器
（1）创建Maven工程flume-interceptor
（2）创建包：com.bigdata_admin.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.bigdata_admin.flume.interceptor包下创建JSONUtils类
package com.bigdata_admin.flume.interceptor;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
public class JSONUtils {
public static boolean isJSONValidate(String log){
try {
JSON.parse(log);
return true;
}catch (JSONException e){
return false;
}
}
}（5）在com.bigdata_admin.flume.interceptor包下创建ETLInterceptor类
package com.bigdata_admin.flume.interceptor;
import com.alibaba.fastjson.JSON;
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 {
@Override
public void initialize() {
}
@Override
public Event intercept(Event event) {
byte[] body = event.getBody();
String log = new String(body, StandardCharsets.UTF_8);
if (JSONUtils.isJSONValidate(log)) {
return event;
} else {
return null;
}
}
@Override
public 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{
@Override
public Interceptor build() {
return new ETLInterceptor();
}
@Override
public void configure(Context context) {
}
}
@Override
public void close() {
}
}（6）打包
flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar
（7）需要先将打好的包放入到hadoop102和hadoop103的/opt/module/flume/lib文件夹下面。
1
.3.3
日志采集Flume测试
1）启动Zookeeper、Kafka集群
2）启动hadoop
102
的日志采集Flume
[bigdata_admin@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
[bigdata_admin@hadoop102 kafka]$ bin/kafka-console-consumer.sh –bootstrap-server hadoop102:9092 –topic topic_log
4
）生成模拟数据
[bigdata_admin@hadoop102 ~]$ lg.sh
5
）观察
Kafka
消费者是否能消费到数据
1
.3.4
日志采集Flume启停脚本
1
）分发日志采集Flume配置文件和拦截器
若上述测试通过，需将hadoop102节点的Flume的配置文件和拦截器jar包，向另一台日志服务器发送一份。
[bigdata_admin@hadoop102 flume]$ scp -r job hadoop103:/opt/module/flume/
[bigdata_admin@hadoop102 flume]$ scp lib/flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar hadoop103:/opt/module/flume/lib/
2）方便起见，此处编写一个日志采集Flume进程的启停脚本
（1）在hadoop102节点的/home/bigdata_admin/bin目录下创建脚本f1.sh
[bigdata_admin@hadoop102 bin]$ vim f1.sh
在脚本中填写如下内容。
#!/bin/bash
case $1 in
"start"){
for i in hadoop102 hadoop103
do
echo " --------启动 $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 hadoop103
do
echo " --------停止 $i 采集flume-------"
ssh $i "ps -ef | grep file_to_kafka | grep -v grep |awk '{print \$2}' | xargs -n1 kill -9 "
done
};;
esac（2）增加脚本执行权限
[bigdata_admin@hadoop102 bin]$ chmod 777 f1.sh
（3）f1启动
[bigdata_admin@hadoop102 module]$ f1.sh
start
（4）f2停止
[bigdata_admin@hadoop102 module]$ f1.sh stop
1
.
4
日志消费Flume
1
.4.1
日志消费Flume配置概述
按照规划，该Flume需将Kafka中topic_log的数据发往HDFS。并且对每天产生的用户行为日志进行区分，将不同天的数据发往HDFS不同天的路径。
此处选择KafkaSource、FileChannel、HDFSSink。
关键配置如下：
1
.4.2
日志消费Flume配置实操
1）创建Flume配置文件
在hadoop104节点的Flume的job目录下创建kafka_to_hdfs_log.conf。
[bigdata_admin@hadoop104 flume]$ vim job/kafka_to_hdfs_log.conf
配置文件内容如下：
## 组件
a1.sources=r1
a1.channels=c1
a1.sinks=k1
## source1
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.batchSize = 5000
a1.sources.r1.batchDurationMillis = 2000
a1.sources.r1.kafka.bootstrap.servers = hadoop102:9092,hadoop103:9092,hadoop104:9092
a1.sources.r1.kafka.topics=topic_log
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.bigdata_admin.flume.interceptor.TimeStampInterceptor$Builder
## channel1
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /opt/module/flume/checkpoint/behavior1
a1.channels.c1.dataDirs = /opt/module/flume/data/behavior1/
a1.channels.c1.maxFileSize = 2146435071
a1.channels.c1.capacity = 1000000
a1.channels.c1.keep-alive = 6
## sink1
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /origin_data/gmall/log/topic_log/%Y-%m-%d
a1.sinks.k1.hdfs.filePrefix = log-
a1.sinks.k1.hdfs.round = false
a1.sinks.k1.hdfs.rollInterval = 10
a1.sinks.k1.hdfs.rollSize = 134217728
a1.sinks.k1.hdfs.rollCount = 0
## 控制输出文件是原生文件。
a1.sinks.k1.hdfs.fileType = CompressedStream
a1.sinks.k1.hdfs.codeC = gzip
## 拼装
a1.sources.r1.channels = c1
a1.sinks.k1.channel= c1
注：配置优化
（1）FileChannel优化
通过配置
dataDirs
指向多个路径
，每个路径对应不同的硬盘，增大Flume吞吐量。
官方说明如下：
Comma
separated list of directories for storing log files. Using multiple directories on separate disks can improve file channel peformance
checkpointDir和backupCheckpointDir也尽量配置在不同硬盘对应的目录中，
保证checkpoint坏掉后，可以快速使用backupCheckpointDir恢复数据。
（2）HDFS Sink优化
①HDFS存入大量小文件，有什么影响？
元数据层面：
每个小文件都有一份元数据，其中包括文件路径，文件名，所有者，所属组，权限，创建时间等，这些信息都保存在Namenode内存中。所以小文件过多，会占用Namenode服务器大量内存，影响Namenode性能和使用寿命。
计算层面：
默认情况下MR会对每个小文件启用一个Map任务计算，非常影响计算性能。同时也影响磁盘寻址时间。
②HDFS小文件处理
官方默认的这三个参数配置写入HDFS后会产生小文件，hdfs.rollInterval、hdfs.rollSize、hdfs.rollCount。
基于以上hdfs.rollInterval=3600，hdfs.rollSize=134217728，hdfs.rollCount =0几个参数综合作用，效果如下：
- 文件在达到128M时会滚动生成新文件
- 文件创建超3600秒时会滚动生成新文件
2
）编写Flume拦截器
（1）在com.bigdata_admin.flume.interceptor包下创建TimeStampInterceptor类
package com.bigdata_admin.flume.interceptor;
import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
public class TimeStampInterceptor implements Interceptor {
private ArrayList<Event> events = new ArrayList<>();
@Override
public void initialize() {
}
@Override
public Event intercept(Event event) {
Map<String, String> headers = event.getHeaders();
String log = new String(event.getBody(), StandardCharsets.UTF_8);
JSONObject jsonObject = JSONObject.parseObject(log);
String ts = jsonObject.getString("ts");
headers.put("timestamp", ts);
return event;
}
@Override
public List<Event> intercept(List<Event> list) {
events.clear();
for (Event event : list) {
events.add(intercept(event));
}
return events;
}
@Override
public void close() {
}
public static class Builder implements Interceptor.Builder {
@Override
public Interceptor build() {
return new TimeStampInterceptor();
}
@Override
public void configure(Context context) {
}
}
}（2）重新打包
flume-interceptor-1.0-SNAPSHOT-jar-with-dependencies.jar
（3）需要先将打好的包放入到hadoop104的/opt/module/flume/lib文件夹下面。
1
.4
.
3
日志消费Flume测试
1）启动Zookeeper、Kafka集群
2）启动日志采集Flume
[bigdata_admin@hadoop102 ~]$ f1.sh start
3）启动hadoop
104
的日志消费Flume
[bigdata_admin@hadoop104 flume]$ bin/flume-ng agent -n a1 -c conf/ -f job/kafka_to_hdfs_log.conf -Dflume.root.logger=info,console
4
）生成模拟数据
[bigdata_admin@hadoop102 ~]$ lg.sh
5
）观察
HDFS
是否出现数据
1
.4
.
4
日志消费Flume启停脚本
若上述测试通过，为方便，此处创建一个Flume的启停脚本。
1
）在
hadoop102
节点的/home/bigdata_admin/bin目录
下创建
脚本
f2.sh
[bigdata_admin@hadoop102 bin]$ vim f2.sh
在脚本中填写如下内容：
#!/bin/bash
case $1 in
"start")
echo " --------启动 hadoop104 日志数据flume-------"
ssh hadoop104 "nohup /opt/module/flume/bin/flume-ng agent -n a1 -c /opt/module/flume/conf -f /opt/module/flume/job/kafka_to_hdfs_log.conf >/dev/null 2>&1 &"
;;
"stop")
echo " --------停止 hadoop104 日志数据flume-------"
ssh hadoop104 "ps -ef | grep kafka_to_hdfs_log | grep -v grep |awk '{print \$2}' | xargs -n1 kill"
;;
esac
2
）增加脚本执行
权限
[bigdata_admin@hadoop102 bin]$ chmod 777 f2.sh
3
）
f2
启动
[bigdata_admin@hadoop102 module]$ f2.sh start
4
）
f2
停止
[bigdata_admin@hadoop102 module]$ f2.sh stop
1
.
5
采集通道启动/停止脚本
1
）在/home/bigdata_admin/bin目录
下创建
脚本
cluster.sh
[bigdata_admin@hadoop102 bin]$ vim cluster.sh
在脚本中填写如下内容：
#!/bin/bash
case $1 in
"start"){
echo ================== 启动 集群 ==================
#启动 Zookeeper集群
zk.sh start
#启动 Hadoop集群
cdh.sh start
#启动 Kafka采集集群
kf.sh start
#启动 Flume采集集群
f1.sh start
#启动 Flume消费集群
f2.sh start
};;
"stop"){
echo ================== 停止 集群 ==================
#停止 Flume消费集群
f2.sh stop
#停止 Flume采集集群
f1.sh stop
#停止 Kafka采集集群
kf.sh stop
#停止 Hadoop集群
cdh.sh stop
#循环直至 Kafka 集群进程全部停止
#xcall.sh 是我们写的脚本，作用是在集群的每个节点都执行一次后面的命令。此处 xcall.sh jps 的作用是查看所有节点的 java 进程
#grep Kafka 的作用是过滤所有 Kafka 进程
#wc -l 是统计行数，每个进程会在 jps 中单独占据一行，因此行数等于进程数
#$()的作用是将括号内命令的执行结果作为值取出来
#因此如下命令的作用是统计集群未停止的 Kafka 进程数然后将进程数赋值给 kafka_count 变量
kafka_count=$(xcall.sh jps | grep Kafka | wc -l)
#判断 kafka_count 变量的值是否大于零，如果是则说明仍有未停止的 Kafka 进程，此时不能停止 Zookeeper，因为 Kafka 的工作要依赖于 Zookeeper 的节点，如果在 Kafka 进程停止之前停止了 Zookeeper，可能会导致本次 Kafka 进程无法正常停止。所以当 Kafka 进程数大于零时进入循环，休眠一秒，然后重新统计 Kafka 进程数，直至 Kafka 进程数为零跳出循环，才能进行下一步（停止 Zookeeper 集群）
while [ $kafka_count -gt 0 ]
do
sleep 1
kafka_count=$( xcall.sh jps | grep Kafka | wc -l)
echo "当前未停止的 Kafka 进程数为 $kafka_count"
done
#停止 Zookeeper集群
zk.sh stop
};;
esac
2
）增加脚本执行
权限
[bigdata_admin@hadoop102 bin]$ chmod u+x cluster.sh
3
）
cluster
集群启动脚本
[bigdata_admin@hadoop102 module]$ cluster.sh start
4
）
cluster
集群停止脚本
[bigdata_admin@hadoop102 module]$ cluster.sh stop