实战1：采集文件内容上传到HDFS
需求
采集目录中已有的文件内容，存储到HDFS
分析
source是要基于目录的，channel建议使用file，可以保证不丢数据，sink使用hdfs
source使用Spooling Directory Source
配置Agent
从example.conf 复制一个新的配置文件
创建数据目录
文件配置信息如图
# Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = spooldir a1.sources.r1.spoolDir = /data/log/studentDir # Use a channel which buffers events in memory a1.channels.c1.type = file a1.channels.c1.checkpointDir = /data/soft/apache-flume-1.9.0-bin/data/studentDir/checkpoint a1.channels.c1.dataDirs = /data/soft/apache-flume-1.9.0-bin/data/studentDir/data # Describe the sink a1.sinks.k1.type = hdfs a1.sinks.k1.hdfs.path = hdfs://192.168.197.101:9000/flume/studentDir a1.sinks.k1.hdfs.filePrefix = stu- 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 # Bind the source and sink to the channel a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1
主要配置checkpointDir和dataDir，
因为这两个目录默认会在用户家目录下生成，
建议修改到其他地方
checkpointDir是存放检查点目录
data是存放数据的目录
最后是sink 因为要向hdfs中输出数据，所以可以使用hdfssink
hdfs.path：是必填项，指定hdfs上的存储目录
filePrefix参数：文件前缀，这个属于可选项
writeFormat：建议改为Text，后期Hive和Impala可以操作这份数据
fileType：默认是SequenceFile，还支持DataStream 和 CompressedStream，DataStream 不会对输出 数据进行压缩，CompressedStream 会对输出数据进行压缩，在这里我们先不使用压缩格式的，所以选 择DataStream
hdfs.rollInterval：默认值是30，单位是秒，表示hdfs多长时间切分一个文件，因为这个采集程序是一 直运行的，只要有新数据，就会被采集到hdfs上面，hdfs默认30秒钟切分出来一个文件，如果设置 为0表示不按时间切文件
hdfs.rollSize：默认是1024，单位是字节，最终hdfs上切出来的文件大小都是1024字节，如果设置为0 表示不按大小切文件
hdfs.rollCount：默认设置为10，表示每隔10条数据切出来一个文件，如果设置为0表示不按数据条数 切文件 这三个参数，如果都设置的有值，哪个条件先满足就按照哪个条件都会执行。 在实际工作中一般会根据时间或者文件大小来切分文件，我们之前在工作中是设置的时间和文件大小 相结合，时间设置的是一小时，文件大小设置的128M，这两个哪个满足执行哪个 所以针对hdfssink的配置最终是这样的。
在启动前，我们初始化一份测试数据
接来下启动Agent
flume-ng agent --name a1 --conf /data/soft/apache-flume-1.9.0-bin/conf/ --conf-file ../conf/file-to-hdfs.conf -Dflume.root.logger=INFO,console
Flume Agent启动成功，成功将数据文件写入HDFS
此时发现文件已经生成了，只不过默认情况下现在的文件是 .tmp 结尾的，表示它在被使用，
因为Flume 只要采集到数据就会向里面写，这个后缀默认是由 hdfs.inUseSuffix 参数来控制的。
文件名上还拼接了一个当前时间戳，这个是默认文件名的格式，当达到文件切割时机的时候会给文件改名 字，去掉.tmp 这个文件现在也是可以查看的，里面的内容其实就是class1.dat文件中的内容。
查看文件内容：表明我们此次采集结果是成功的
那Flume怎么知道哪些文件是新文件呢？它会不会重复读取同一个文件的数据呢？
我们进入目录：/data/log/studentDir
我们发现此时这个文件已经被加了一个后缀 .COMPLETED ，表示这个文件已经被读取过了，所以Flume在 读取的时候会忽略后缀为 .COMPLETED 的文件
接着我们再看一下channel中的数据，因为数据是存在本地磁盘文件中的，所以是可以去看一下的，进入 dataDir指定的目录
发现里面有一个 log-1 的文件，这个文件中存储的其实就是读取到的内容,不过在这无法直接查看
现在我们想看一下Flume最终生成的文件是什么样子的，难道要根据配置等待1个小时或者弄一个128M 的文件过来吗， 其实也没必要，我们可以暴力操作一下 停止Agent就可以看到了，当Agent停止的时候就会去掉 .tmp 标志了
那我再重启Agent之后，会不会再给加上.tmp呢，不会了，每次停止之前都会把所有的文件解除占用状 态，下次启动的时候如果有新数据，则会产生新的文件，这其实就模拟了一下自动切文件之后的效果。
实战2：日志数据汇总采集
需求
- 将bigdata02和bigdata03机器实时产生的日志数据汇总到bigdata04中
- 通过bigdata04将数据输出到HDFS指定目录
这里注意：HDFS目录要按天生产每天一个目录。
分析
图解：
这里需要用到3个Agent
- Agent1负责采集机器bigdata02数据
- Agent2负责采集机器bigdata03数据
- Agent3负责汇总机器1和2数据到机器3再统一输出到HDFS
- Agent1和Agent2因为要实时读取文件中新增数据，所以使用基于文件的source，Exec Source。
- Channel统一使用基于内存的Channel-Memory Channel
- 由于需要汇总数据，所以sink端加快传输使用Avro Sink
- 备注：Avro是一种序列化的手段，经过序列化的数据进行传输的时候效率非常高，Avro Sink发送的数据可以直接被Avro Source接受，无缝衔接
实战
以下定义02为A、03为B、04为C
首先在02机器上配置Flume
配置Agent，创建文件 file-to-avro-104.conf
在03机器上配置Flume
与02机器一样的操作
配置Agent，创建文件file-to-avro-104.conf
在04机器上配置文件avro-to-hdfs.conf
这里有个注意的点：
在指定Agent中sink配置的时候注意，我们的需求是需要按天在hdfs中创建目录，并把当天的数据上传到 当天的日期目录中，这也就意味着hdfssink中的path不能写死，需要使用变量，动态获取时间，查看官 方文档可知，在hdfs的目录中需要使用%Y%m%d。
这个时间其实是需要从数据里面抽取，咱们前面 说过数据的基本单位是Event，Event是一个对象，后面我们会详细分析，在这里大家先知道它里面包含 的既有我们采集到的原始的数据，还有一个header属性，这个header属性是一个key-value结构的，我 们现在抽取时间就需要到event的header中抽取，但是默认情况下event的header中是没有日期的，强行 抽取是会报错的，会提示抽取不到，返回空指针异常。
其实官方文档中也说了，可以使用hdfs.useLocalTimeStamp或者时间 拦截器，暂时最简单直接的方式就是使用hdfs.useLocalTimeStamp，这个属 性的值默认为false，需要改为true
三台机器中的Flume Agent都配置好了，在开始启动之前需要先在bigdata02和bigdata03中生成测试数 据，为了模拟真实情况，在这里我们就开发一个脚本，定时向文件中写数据。
#!/bin/bash # 循环向文件中生成数据 while [ "1" = "1" ] do # 获取当前时间戳 curr_time=`date +%s` # 获取当前主机名 name=`hostname` echo ${name}_${curr_time} >> /data/log/access.log # 暂停1秒 sleep 1 done
1.首先在bigdata02上创建/data/log目录，然后创建 generateAccessLog.sh 脚本
2.接着在bigdata03上创建/data/log目录，然后创建 generateAccessLog.sh 脚本
3.接下来开始启动相关的服务进程 首先启动bigdata04上的agent服务
这里要注意下启动顺序
首先应该启动的是04机器、如果没有启动04就启动了02和03，会丢失一部分数据
- 启动04
../bin/flume-ng agent --name a1 --conf /data/soft/apache-flume-1.9.0-bin/conf/ --conf-file avro-to-hdfs.conf -Dflume.root.logger=INFO,console
- 启动03
../bin/flume-ng agent --name a1 --conf /data/soft/apache-flume-1.9.0-bin/conf/ --conf-file file-to-avro-104.conf -Dflume.root.logger=INFO,console
初始化测试数据
sh -x generateAccessLog.sh
- 启动02
../bin/flume-ng agent --name a1 --conf /data/soft/apache-flume-1.9.0-bin/conf/ --conf-file file-to-avro-104.conf -Dflume.root.logger=INFO,console
初始化测试数据
sh -x generateAccessLog.sh
验证数据结果
启动之后稍等一会就可以看到数据了，我们观察数据的变化，会发现hdfs中数据增长的不 是很快，它会每隔一段时间添加一批数据，实时性好像没那么高
注意
这是因为avrosink中有一个配置batch-size，它的默认值是100，也就是每次发送100条数据，如果数据 不够100条，则不发送。 具体这个值设置多少合适，要看你source数据源大致每秒产生多少数据，以及你希望的延迟要达到什么 程度，如果这个值设置太小的话，会造成sink频繁向外面写数据，这样也会影响性能。
实战结束
最终，依次停止bigdata02、bigdata03中的服务，最后停止bigdata04中的服务
