大数据开发工程师-第七周 第2章 极速上手Flume使用 采集网络日志上传到HDFS

---

第七周 第2章 极速上手Flume使用 采集网络日志上传到HDFS

前面我们讲了两个案例的使用，接下来看一个稍微复杂一点的案例：
需求是这样的，
1. 将A和B两台机器实时产生的日志数据汇总到机器C中
2. 通过机器C将数据统一上传至HDFS的指定目录中

注意：HDFS中的目录是按天生成的，每天一个目录
看下面这个图，来详细分析一下

根据刚才的需求分析可知，我们一共需要三台机器
这里使用bigdata02和bigdata03采集当前机器上产生的实时日志数据，统一汇总到bigdata04机器上。

  其中bigdata02和bigdata03中的source使用基于file的source，ExecSource，因为要实时读取文件中的新增数据
  channel在这里我们使用基于内存的channel，因为这里是采集网站的访问日志，就算丢一两条数据对整体结果影响也不大，我们只希望采集到的数据可以快读进入hdfs中，所以就选择了基于内存的channel。

由于bigdata02和bigdata03的数据需要快速发送到bigdata04中，为了快速发送我们可以通过网络直接传输，sink建议使用avrosink，avro是一种数据序列化系统，经过它序列化的数据传输起来效率更高，并且它对应的还有一个avrosource，avrosink的数据可以直接发送给avrosource，所以他们可以无缝衔接。

这样bigdata04的source就确定了 使用avrosource、channel还是基于内存的channel，sink就使用hdfssink，因为是要向hdfs中写数据的。

这里面的组件，只有execsource、avrosource、avrosink我们还没有使用过，其他的组件都使用过了。

最终需要在每台机器上启动一个agent，启动的时候需要注意先后顺序，先启动bigdata04上面的，再启动bigdata02和bigdata03上面的。

配置环境

bigdata02

1：在bigdata02上安装Flume并配置Agent
上传Flume的安装包，解压

配置Agent，创建文件 file-to-avro-101.conf

bigdata03

2：在bigdata03上安装Flume并配置Agent
上传Flume的安装包，解压

配置Agent，创建文件file-to-avro-102.conf

配置文件

注意：bigdata02和bigdata03中配置的a1.sinks.k1.port 的值45454需要和bigdata04中配置的一致

bigdata02 conf

[root@bigdata02 conf]# vim file-to-avro-101.conf

a1.sources=r1
a1.channels=c1
a1.sinks=k1

a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /data/log/networkLogUploadToHdfsExample/access.log


a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100


a1.sinks.k1.type = avro
# ip也行
a1.sinks.k1.hostname = bigdata04
# 端口没用过的就行
a1.sinks.k1.port = 45454



a1.sources.r1.channels=c1
a1.sinks.k1.channel = c1

bigdata03 conf

[root@bigdata03 conf]# vim file-to-avro-102.conf

a1.sources=r1
a1.channels=c1
a1.sinks=k1

a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /data/log/networkLogUploadToHdfsExample/access.log


a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100


a1.sinks.k1.type = avro
# ip也行
a1.sinks.k1.hostname = bigdata04
# 端口没用过的就行
a1.sinks.k1.port = 45454

a1.sources.r1.channels=c1
a1.sinks.k1.channel = c1

bigdata04 conf

这台机器我们已经安装过Flume了，所以直接配置Agent即可

在指定Agent中sink配置的时候注意，我们的需求是需要按天在hdfs中创建目录，并把当天的数据上传到当天的日期目录中，这也就意味着hdfssink中的path不能写死，需要使用变量，动态获取时间，查看官方文档可知，在hdfs的目录中需要使用%Y%m%d

在这还有一点需要注意的，因为我们这里需要抽取时间，这个时间其实是需要从数据里面抽取，咱们前面说过数据的基本单位是Event，Event是一个对象，后面我们会详细分析，在这里大家先知道它里面包含的既有我们采集到的原始的数据，还有一个header属性，这个header属性是一个key-value结构的，我们现在抽取时间就需要到event的header中抽取，但是默认情况下event的header中是没有日期的，强行抽取是会报错的，会提示抽取不到，返回空指针异常。

java.lang.NullPointerException: Expected timestamp in the Flume event headers, but it was n

那如何向header中添加日期呢？ 其实官方文档中也说了，可以使用hdfs.useLocalTimeStamp或者时间拦截器，时间拦截器我们后面会讲，暂时最简单直接的方式就是使用hdfs.useLocalTimeStamp，这个属性的值默认为false，需要改为true。

[root@bigdata04 conf]# vim avro-to-hdfs.conf 
 

a1.sources=r1
a1.channels=c1
a1.sinks=k1


a1.sources.r1.type = avro
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 45454(一定要和bigdata02,03的avro sink里配置的端口号一致)

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://bigdata04:9000/flume/networkLogUploadToHdfsExaple/%Y%m%d
a1.sinks.k1.hdfs.filePrefix = access-
a1.sinks.k1.hdfs.fileSuffix=.log
a1.sinks.k1.hdfs.fileType=DataStream
a1.sinks.k1.hdfs.writeFormat=Text
a1.sinks.k1.hdfs.rollInterval=3600
a1.sinks.k1.hdfs.rollSize=134217728
a1.sinks.k1.hdfs.rollCount=0
#(不设置前面使用时间变量会报错)使用event header里的timestamp要麻烦些
a1.sinks.k1.hdfs.useLocalTimeStamp=true

a1.sources.r1.channels=c1
a1.sinks.k1.channel=c1

模拟数据源

三台机器中的Flume Agent都配置好了，在开始启动之前需要先在bigdata02和bigdata03中生成测试数据，为了模拟真实情况，在这里我们就开发一个脚本，定时向文件中写数据

在bigdata02和bigdata03中使用这个脚本生成数据
首先在bigdata02上创建/data/log目录，然后创建 generateAccessLog.sh 脚本

bigdata02

[root@bigdata02 log]# vim SimulateData.sh 

#!/bin/bash
# 循环向文件中生成数据
while [ "1"="1" ]
do
        # 获取当前时间戳
        curr_time=`date +%s`
        # 获取当前主机名
        name=`hostname`
        echo ${name}_${curr_time} >> /data/log/networkLogUploadToHdfsExample/access.log
        sleep 1
done

bigdata03

接着在bigdata03上创建/data/log目录，然后创建 generateAccessLog.sh 脚本

同上

启动进程

接下来开始启动相关的服务进程
首先启动bigdata04上的agent服务
接下来启动bigdata-02上的agent服务和shell脚本
最后启动bigdata-03上的agent服务和shell脚本

启动bigdata04

1 [root@bigdata04 apache-flume-1.9.0-bin]# bin/flume-ng agent --name a1 --conf-file conf/avro-to-hdfs.conf -Dflume.root.logger=INFO,console

启动bigdata02

1 [root@bigdata04 apache-flume-1.9.0-bin]# bin/flume-ng agent --name a1 --conf-file conf/file-to-avro.conf -Dflume.root.logger=INFO,console

启动bigdata03

1 [root@bigdata04 apache-flume-1.9.0-bin]# bin/flume-ng agent --name a1 --conf-file conf/file-to-avro.conf -Dflume.root.logger=INFO,console

注意

1.shell脚本
while [ 1=1 ] 
# 两个空格不能少

2.flume配置
注释不要写在语句的同一行后面

3.启动agent时
先启动bigdata02,bigdata03和先关bigdata04都会造成数据流失

结果查看

注意：启动之后稍等一会就可以看到数据了，我们观察数据的变化，会发现hdfs中数据增长的不是很快，它会每隔一段时间添加一批数据，实时性好像没那么高？
这是因为avrosink中有一个配置batch-size，它的默认值是100，也就是每次发送100条数据，如果数据不够100条，则不发送。
具体这个值设置多少合适，要看你source数据源大致每秒产生多少数据，以及你希望的延迟要达到什么程度，如果这个值设置太小的话，会造成sink频繁向外面写数据，这样也会影响性能。

最终，依次停止bigdata02、bigdata03中的服务，最后停止bigdata04中的服务

---