Flume(一)

谁借莪１个温暖的怀抱￠ 2023-06-30 09:25 30阅读 0赞

## [https://www.cnblogs.com/xuziyu/p/11004103.html][https_www.cnblogs.com_xuziyu_p_11004103.html] ##

## Flume的定义 ##

*  Flume是一个分布式的、高可靠的、高可用的将大批量的不同数据源的日志数据收集、聚合、移动到数据中心（HDFS）进行存储的系统。即是日志采集和汇总的工具
 *  Logstash、FileBeat是ES栈的日志数据抽取工具，他们和Flume很类似，前者是轻量级、后者是重量级，若项目组使用的是ES栈技术，那完全可以使用Logstash取代Flume。

## 版本 ##

![format_png][]

*  NG: 1.x的版本   (N=NEW)
 *  OG：0.9.x的版本,不用管(O=OLD)
 *  由于我使用的是CDH5.7.0，故选择flume-ng-1.6.0-cdh5.7.0版本，注意此1.6和社区的1.6有差别。

*  ### flume的优势： ###

1.  可以高速采集数据，采集的数据能够以想要的文件格式及压缩方式存储在hdfs上
2.  事务功能保证了数据在采集的过程中数据不丢失
3.  部分Source保证了Flume挂了以后重启依旧能够继续在上一次采集点采集数据，真正做到数据零丢失

*  ### flume的组成 ###
 *  flume有3大组件

1.  source(源端数据采集)：Flume提供了各种各样的Source、同时还提供了自定义的Source
2.  Channel(临时存储聚合数据)：主要用的是memory channel和File channel（生产最常用），生产中channel的数据一定是要监控的，防止sink挂了，撑爆channel
3.  Sink（移动数据到目标端）：如HDFS、KAFKA、DB以及自定义的sink

*  ### flume的架构 ###
 *  flume的agent架构
 *  单Agent：

![format_png 1][]

*  串联Agent：

![format_png 2][]

*  并联Agent(生产中最多的使用)：

![format_png 3][]

*   多sinkAgent(也很常见)：

![format_png 4][]

##  flume部署 ##

*  ### 打开官网http://archive.cloudera.com/cdh5/cdh/5/ ###

![format_png 5][]

*  ### 上传并解压 ###

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 </td> 
 <td> <code>[hadoop@hadoop001 app]$ rz</code> &nbsp; <code>flume-ng-1.6.0-cdh5.7.0.tar.gz</code> &nbsp; <code>[hadoop@hadoop001 app]$tar -xzvf&nbsp; flume-ng-1.6.0-cdh5.7.0.tar.gz</code> </td> 
 </tr> 
 </tbody> 
</table>

\#修改配置文件，添加JAVA\_HOME

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 </td> 
 <td> <code>[hadoop@hadoop001 app]$ cd ~/app/apache-flume-1.6.0-cdh5.7.0-bin</code> <code>[hadoop@hadoop001 apache-flume-1.6.0-cdh5.7.0-bin]$ cp ~/app/apache-flume-1.6.0-cdh5.7.0-bin/conf/flume-env.sh.template ~/app/apache-flume-1.6.0-cdh5.7.0-bin/conf/flume-env.sh</code> <code>[hadoop@hadoop001 apache-flume-1.6.0-cdh5.7.0-bin]$ vim ~/app/apache-flume-1.6.0-cdh5.7.0-bin/conf/flume-env.sh</code> <code>export JAVA_HOME=/usr/java/jdk1.8.0_45</code> </td> 
 </tr> 
 </tbody> 
</table>

\#添加环境变量

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 </td> 
 <td> <code>hadoop@hadoop001 bin]$ soruce ~/.bash_profile</code> <code>export FLUME_HOME=/home/hadoop/app/apache-flume-1.6.0-cdh5.7.0-bin</code> <code>export PATH=$FLUME_HOME/bin:$PATH</code> <code>[hadoop@hadoop001 bin]$ source ~/.bash_profile</code> <code>[hadoop@hadoop001 bin]$ which flume-ng</code> <code>~/app/apache-flume-1.6.0-cdh5.7.0-bin/bin/flume-ng</code> </td> 
 </tr> 
 </tbody> 
</table>

*  ### Agent配置使用案列 ###

1.  Flume的使用其实就是Source、Channel、Sink的配置
2.  Agent=Source+Channel+Sink，其实agent就是Flume的配置文件
3.  一个配置文件可以配置多个Agent的。
4.  Event：Flume数据传输的最小单位，一个EVent就是一条记录，由head和body两个部分组成，head存储的是管道，body存储的是字节数组

*  ### Flume文件配置 ###

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

[hadoop@hadoop001 conf]$ vim  /home/hadoop/app/apache-flume-1.6.0-cdh5.7.0-bin/conf/example.conf
    
    # example.conf: A single-node Flume configuration
    
    # Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 44444
    
    # Describe the sink
    a1.sinks.k1.type = logger
    
    # Use a channel which buffers events in memory
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    # Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

*   **NetSource:黑色的配置是必填项**

![format_png 6][]

*  **memory channel：capatity=>channel的存储最大event(消息)个数，生产至少10万条，**
 *  **transationCapacity=>最多达到多少条必须提交事务。生产也必须调大。**

![format_png 7][]

*  **logger：就是控制台类型的sink**
 *  **注意1：一个source可以绑定多个channel,但是一个sink只能绑定一个Channel**

*  ### 启动Agent以及测试 ###
 *  启动

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 </td> 
 <td> <code>#最后一行是为了方便观察输出INFO日志到控制台，可以去掉</code> <code>flume-ng agent \</code> <code>--name a1 \</code> <code>--conf $FLUME_HOME/conf \</code> <code>--conf-file $FLUME_HOME/conf/example.conf \</code> <code>-Dflume.root.logger=INFO,console</code> &nbsp; &nbsp; <code>使用telnet测试：</code> &nbsp; <code>[hadoop@hadoop001 ~]$ telnet localhost 44444</code> <code>Trying 127.0.0.1...</code> <code>Connected to localhost.</code> <code>Escape character&nbsp;</code><code>is</code>&nbsp;<code>'^]'</code><code>.</code> <code>hello</code> <code>OK</code> &nbsp; <code>&nbsp;</code><code>&lt;br&gt;</code> </td> 
 </tr> 
 </tbody> 
</table>

### Exec Source采集文件数据到hdfs ###

生产的架构是： log数据=》flume=》hdfs，这里我们采用简单的Exec Source通过tail -F 数据文件进行数据采集。

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 </td> 
 <td> <code># example.conf: A single-node Flume configuration</code> &nbsp; <code># Name the components on this agent</code> <code>exec-hdfs-agent.sources = exec-source</code> <code>exec-hdfs-agent.sinks = hdfs-sink</code> <code>exec-hdfs-agent.channels = memory-channel</code> &nbsp; <code># Describe/configure the source</code> <code>exec-hdfs-agent.sources.exec-source.type = exec</code> <code>exec-hdfs-agent.sources.exec-source.command = tail -F /home/hadoop/data/access_10000.log</code> <code>exec-hdfs-agent.sources.exec-source.shell = /bin/sh -c</code> &nbsp; <code># Describe the sink</code> <code>exec-hdfs-agent.sinks.hdfs-sink.type = hdfs</code> <code>exec-hdfs-agent.sinks.hdfs-sink.hdfs.path = hdfs:</code><code>//hadoop001:9000/flume/exec</code> <code>exec-hdfs-agent.sinks.hdfs-sink.hdfs.fileType = DataStream</code> <code>exec-hdfs-agent.sinks.hdfs-sink.hdfs.writeFormat = Text</code> &nbsp; <code># Use a channel which buffers events in memory</code> <code>exec-hdfs-agent.channels.memory-channel.type = memory</code> <code>exec-hdfs-agent.channels.memory-channel.capacity = 1000</code> <code>exec-hdfs-agent.channels.memory-channel.transactionCapacity = 100</code> &nbsp; <code># Bind the source and sink to the channel</code> <code>exec-hdfs-agent.sources.exec-source.channels = memory-channel</code> <code>exec-hdfs-agent.sinks.hdfs-sink.channel = memory-channel</code> </td> 
 </tr> 
 </tbody> 
</table>

*  **写hdfs文件时先生成创建一个后缀名称为.tmp的文件，当写完成时，去掉了.tmp**

![format_png 8][]

![format_png 9][]

*  缺点：

1.  虽然此种tail方式可以将日志数据采集到hdfs，但是tail -F进程挂了咋办，还是会丢数据！生产上是行不通的。无法做到高可用。
2.  其次上面的采集流程并未解决生成大量小文件的问题，无法做到高可靠
3.  tail只能监控一个文件，生产中更多的是监控一个文件夹。不能满足需求

*  ### 使用Spooling Directory Source采集文件夹数据到hdfs ###

*  写到HDFS上的文件大小最好是100M左右，比blocksize的值（128M）略低
 *  一般使用rolllnterval（时间）、rollSize（大小）来控制文件的生成，哪个先触发就会生成HDFS文件，将根据条数的roll关闭。
 *  rollSize控制的大小是指的压缩前的，所以若hdfs文件使用了压缩，需调大rollsize的大小
 *  当文件夹下的某个文件被采集到hdfs上，会有个。complete的标志
 *  使用Spooling Directory Source采集文件数据时若该文件数据已经被采集，再对该文件做修改是会报错的停止的，其次若放进去一个已经完成采集的同名数据文件也是会报错停止的
 *  写HDFS数据可按照时间分区，注意改时间刻度内无数据则不会生成该时间文件夹
 *  生成的文件名称默认是前缀+时间戳，这个是可以更改的。

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 </td> 
 <td> <code># example.conf: A single-node Flume configuration</code> &nbsp; <code># Name the components on this agent</code> <code>spool-hdfs-agent.sources = spool-source</code> <code>spool-hdfs-agent.sinks = hdfs-sink</code> <code>spool-hdfs-agent.channels = memory-channel</code> &nbsp; <code># Describe/configure the source</code> <code>spool-hdfs-agent.sources.spool-source.type = spooldir</code> <code>spool-hdfs-agent.sources.spool-source.spoolDir = /home/hadoop/data/flume/spool/input</code> &nbsp; <code># Describe the sink</code> <code>spool-hdfs-agent.sinks.hdfs-sink.type = hdfs</code> <code>spool-hdfs-agent.sinks.hdfs-sink.hdfs.path = hdfs:</code><code>//hadoop001:9000/flume/spool/%Y%m%d%H%M</code> <code>spool-hdfs-agent.sinks.hdfs-sink.hdfs.useLocalTimeStamp =&nbsp;</code><code>true</code> <code>spool-hdfs-agent.sinks.hdfs-sink.hdfs.fileType = CompressedStream</code> <code>spool-hdfs-agent.sinks.hdfs-sink.hdfs.writeFormat = Text</code> <code>spool-hdfs-agent.sinks.hdfs-sink.hdfs.codeC = gzip</code> <code>spool-hdfs-agent.sinks.hdfs-sink.hdfs.filePrefix = wsk</code> <code>spool-hdfs-agent.sinks.hdfs-sink.hdfs.rollInterval = 30</code> <code>spool-hdfs-agent.sinks.hdfs-sink.hdfs.rollSize = 100000000</code> <code>spool-hdfs-agent.sinks.hdfs-sink.hdfs.rollCount = 0</code> &nbsp; <code># Use a channel which buffers events in memory</code> <code>spool-hdfs-agent.channels.memory-channel.type = memory</code> <code>spool-hdfs-agent.channels.memory-channel.capacity = 1000</code> <code>spool-hdfs-agent.channels.memory-channel.transactionCapacity = 100</code> &nbsp; <code># Bind the source and sink to the channel</code> <code>spool-hdfs-agent.sources.spool-source.channels = memory-channel</code> <code>spool-hdfs-agent.sinks.hdfs-sink.channel = memory-channel</code> </td> 
 </tr> 
 </tbody> 
</table>

上述的Spooling Directory Source配置虽然解决了小文件过多以及监控多个文件的问题，但是依旧有如下问题。

*  问题1：虽然能监控一个文件夹，但是无法监控递归的文件夹中的数据
 *  问题2：若采集时Flume挂了，无法保证重启时还从之前文件读取的那一行继续采集数据

基于以上两个问题，此凡是生产也是不可接受的

*  ### (生产版本)使用Taildir Source采集文件夹数据到hdfs ###

1.  Taildir Source 是Apache flume1.7新推出的，但是CDH Flume1.6做了集成
2.  Taildir Source是高可靠（reliable）的source，他会实时的将文件偏移量写到json文件中并保存到磁盘。下次重启Flume时会读取Json文件获取文件O偏移量，然后从之前的位置读取数据，保证数据零丢失
3.  taildir Source可同时监控多个文件夹以及文件。即使文件在实时写入数据。
4.  Taildir Source也是无法采集递归文件下的数据，这需要改造源码
5.  Taildir Source监控一个文件夹下的所有文件一定要用.\*正则

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 </td> 
 <td> <code># example.conf: A single-node Flume configuration</code> &nbsp; <code># Name the components on this agent</code> <code>taildir-hdfs-agent.sources = taildir-source</code> <code>taildir-hdfs-agent.sinks = hdfs-sink</code> <code>taildir-hdfs-agent.channels = memory-channel</code> &nbsp; <code># Describe/configure the source</code> <code>taildir-hdfs-agent.sources.taildir-source.type = TAILDIR</code> <code>taildir-hdfs-agent.sources.taildir-source.filegroups = f1</code> <code>taildir-hdfs-agent.sources.taildir-source.filegroups.f1 = /home/hadoop/data/flume/taildir/input/.*</code> <code>taildir-hdfs-agent.sources.taildir-source.positionFile = /home/hadoop/data/flume/taildir/taildir_position/taildir_position.json</code> &nbsp; <code># Describe the sink</code> <code>taildir-hdfs-agent.sinks.hdfs-sink.type = hdfs</code> <code>taildir-hdfs-agent.sinks.hdfs-sink.hdfs.path = hdfs:</code><code>//hadoop001:9000/flume/taildir/%Y%m%d%H%M</code> <code>taildir-hdfs-agent.sinks.hdfs-sink.hdfs.useLocalTimeStamp =&nbsp;</code><code>true</code> <code>taildir-hdfs-agent.sinks.hdfs-sink.hdfs.fileType = CompressedStream</code> <code>taildir-hdfs-agent.sinks.hdfs-sink.hdfs.writeFormat = Text</code> <code>taildir-hdfs-agent.sinks.hdfs-sink.hdfs.codeC = gzip</code> <code>taildir-hdfs-agent.sinks.hdfs-sink.hdfs.filePrefix = wsk</code> <code>taildir-hdfs-agent.sinks.hdfs-sink.hdfs.rollInterval = 30</code> <code>taildir-hdfs-agent.sinks.hdfs-sink.hdfs.rollSize = 100000000</code> <code>taildir-hdfs-agent.sinks.hdfs-sink.hdfs.rollCount = 0</code> &nbsp; <code># Use a channel which buffers events in memory</code> <code>taildir-hdfs-agent.channels.memory-channel.type = memory</code> <code>taildir-hdfs-agent.channels.memory-channel.capacity = 1000</code> <code>taildir-hdfs-agent.channels.memory-channel.transactionCapacity = 100</code> &nbsp; <code># Bind the source and sink to the channel</code> <code>taildir-hdfs-agent.sources.taildir-source.channels = memory-channel</code> <code>taildir-hdfs-agent.sinks.hdfs-sink.channel = memory-channel</code> </td> 
 </tr> 
 </tbody> 
</table>

[https_www.cnblogs.com_xuziyu_p_11004103.html]: https://www.cnblogs.com/xuziyu/p/11004103.html
[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTMyNjkzNy8yMDE5MDYvMTMyNjkzNy0yMDE5MDYxMDE2MTM1OTM5NC0xMjEzNzk1ODQwLnBuZw?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTMyNjkzNy8yMDE5MDYvMTMyNjkzNy0yMDE5MDYxMDE3MjYxOTQzNi0zMDc5MjcxMzYucG5n?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTMyNjkzNy8yMDE5MDYvMTMyNjkzNy0yMDE5MDYxMDE3MjY1NjYxNC05NDE5MTA3NDYucG5n?x-oss-process=image/format,png
[format_png 3]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTMyNjkzNy8yMDE5MDYvMTMyNjkzNy0yMDE5MDYxMDE3MjcyMDA0Mi05NDMxMzUxMDgucG5n?x-oss-process=image/format,png
[format_png 4]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTMyNjkzNy8yMDE5MDYvMTMyNjkzNy0yMDE5MDYxMDE3MzAwNzEzNi03NDU0NDg5MTYucG5n?x-oss-process=image/format,png
[format_png 5]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTMyNjkzNy8yMDE5MDYvMTMyNjkzNy0yMDE5MDYxMDE3MzQzMDQxMi0xMjE2NjIxNTc4LnBuZw?x-oss-process=image/format,png
[aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg
[format_png 6]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTMyNjkzNy8yMDE5MDYvMTMyNjkzNy0yMDE5MDYxMTEwMDkwMzg0MC02MzQxNTI4MDcucG5n?x-oss-process=image/format,png
[format_png 7]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTMyNjkzNy8yMDE5MDYvMTMyNjkzNy0yMDE5MDYxMTEwMTIwNjUzNC0xMjg3NjkxNTMucG5n?x-oss-process=image/format,png
[format_png 8]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTMyNjkzNy8yMDE5MDYvMTMyNjkzNy0yMDE5MDYxMTE1MDEyOTcwNy0xOTg1NDk5NDQyLnBuZw?x-oss-process=image/format,png
[format_png 9]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTMyNjkzNy8yMDE5MDYvMTMyNjkzNy0yMDE5MDYxMTE1MDE1MzczMC02ODk5ODA4Ni5wbmc?x-oss-process=image/format,png