ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 数据库 > 数据库开发技术 > Flume安装及配置

Flume安装及配置

原创 数据库开发技术 作者：HelloTech技术派 时间：2018-04-24 08:01:00 0 删除 编辑

Flume 提供了大量内置的 Source、Channel 和 Sink 类型。而且不同类型的 Source、Channel 和 Sink 可以自由组合—–组合方式基于配置文件的设置，非常灵活。比如：Channel 可以把事件暂存在内存里，也可以持久化到本地硬盘上。Sink 可以把日志写入 HDFS、HBase，甚至是另外一个 Source 等。

安装

  • 下载源码包到 /usr/local/src 目录下

  • 解压tar.gz包

  • 修改配置文件，在 /usr/local/src/apache-flume-1.7.0-bin/conf 目录下

  • 配置环境变量

核心配置

  • 在 /usr/local/src/apache-flume-1.7.0-bin/conf 目录下提供默认配置模板

  • 配置示例——NetCat Source：监听一个指定的网络端口，只要应用程序向这个端口里面写数据，这个 Source 组件就可以获取到数据。 其中 Sink：logger， Channel：memory 。

  • 配置如下：

    

  1. # Name the components on this agent
    2. 

  2. agent.sources = netcat
    3. 

  3. agent.channels = memoryChannel
    4. 

  4. agent.sinks = loggerSink
    5. 

    6. 

  6. # configure the source
    7. 

  7. agent.sources.netcat.type = netcat
    8. 

  8. agent.sources.netcat.bind = 192.168.111.238
    9. 

  9. agent.sources.netcat.port = 8888
    10. 

    11. 

  11. # Describe the sink
    12. 

  12. agent.sinks.loggerSink.type = logger
    13. 

    14. 

  14. # Each channel's type is defined.
    15. 

  15. agent.channels.memoryChannel.type = memory
    16. 

  16. # It specifies the capacity of the memory channel
    17. 

  17. agent.channels.memoryChannel.capacity = 100
    18. 

    19. 

  19. # Bind the source and sink to the channel
    20. 

  20. agent.sources.netcat.channels = memoryChannel
    21. 

  21. agent.sinks.loggerSink.channel = memoryChannel
    22.

运行测试

  • 启动名字为 agent 的 Flume Agent 服务端进程

  1.  ./bin/flume-ng agent -n agent -c ./conf -f ./conf/flume-netcat.properties -Dflume.root.logger=DEBUG,console

参数说明：

-n 指定 agent 名称(与配置文件中代理的名字相同)

-c 指定 Flume 中配置文件的目录

-f 指定配置文件

-Dflume.root.logger=DEBUG,console 设置日志等级

  • 使用 telnet 发送数据

    

  1. zhwye@ubuntu238:~$ telnet 192.168.111.238 8888
    2. 

  2. Trying 192.168.111.238...
    3. 

  3. Connected to 192.168.111.238.
    4. 

  4. Escape character is '^]'.
    5. 

  5. big data world
    6. 

  6. OK
    7.

  • 服务端进程收集到的日志数据如下

    

  1. ...
    2. 

  2. 2018-04-12 07:07:22,965 (netcat-handler-0) [DEBUG - org.apache.flume.source.NetcatSource$NetcatSocketHandler.run(NetcatSource.java:315)] Starting connection handler
    3. 

  3. 2018-04-12 07:07:31,457 (netcat-handler-0) [DEBUG - org.apache.flume.source.NetcatSource$NetcatSocketHandler.run(NetcatSource.java:327)] Chars read = 16
    4. 

  4. 2018-04-12 07:07:31,468 (netcat-handler-0) [DEBUG - org.apache.flume.source.NetcatSource$NetcatSocketHandler.run(NetcatSource.java:331)] Events processed = 1
    5. 

  5. 2018-04-12 07:07:35,191 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 62 69 67 20 64 61 74 61 20 77 6F 72 6C 64 0D    big data world. }
    6.

常用配置模式

NetCat Source——HDFS Sink 配置

  • 监听一个指定的网络端口，只要应用程序向这个端口里面写数据，这个 Source 组件就可以获取到数据。 其中 Sink：HDFS， Channel：file 

  • 配置文件

    

  1. # Name the components on this agent
    2. 

  2. a1.sources = r1
    3. 

  3. a1.sinks = k1
    4. 

  4. a1.channels = c1
    5. 

    6. 

  6. # Describe/configure the source
    7. 

  7. a1.sources.r1.type = netcat
    8. 

  8. a1.sources.r1.bind = 192.168.111.238
    9. 

  9. a1.sources.r1.port = 8888
    10. 

    11. 

  11. # Describe the sink
    12. 

  12. a1.sinks.k1.type = hdfs
    13. 

  13. a1.sinks.k1.hdfs.path = hdfs://hadoop238:9000/dataoutput
    14. 

  14. a1.sinks.k1.hdfs.writeFormat = Text
    15. 

  15. a1.sinks.k1.hdfs.fileType = DataStream
    16. 

  16. a1.sinks.k1.hdfs.rollInterval = 10
    17. 

  17. a1.sinks.k1.hdfs.rollSize = 0
    18. 

  18. a1.sinks.k1.hdfs.rollCount = 0
    19. 

  19. a1.sinks.k1.hdfs.filePrefix = %Y-%m-%d-%H-%M-%S
    20. 

  20. a1.sinks.k1.hdfs.useLocalTimeStamp = true
    21. 

    22. 

  22. # Use a channel which buffers events in file
    23. 

  23. a1.channels.c1.type = file
    24. 

  24. a1.channels.c1.checkpointDir = /usr/flume/checkpoint
    25. 

  25. a1.channels.c1.dataDirs = /usr/flume/data
    26. 

    27. 

  27. # Bind the source and sink to the channel
    28. 

  28. a1.sources.r1.channels = c1
    29. 

  29. a1.sinks.k1.channel = c1
    30.

  • 启动 Flume agent a1 服务端

  1. flume-ng  agent -n a1  -c ../conf  -f ../conf/flume-hdfs.conf   -Dflume.root.logger=DEBUG,console

  • 使用 telnet 发送数据

    

  1. zhwye@ubuntu238:~$ telnet 192.168.111.238 8888
    2. 

  2. Trying 192.168.111.238...
    3. 

  3. Connected to 192.168.111.238.
    4. 

  4. Escape character is '^]'.
    5. 

  5. big data world
    6. 

  6. OK
    7.

Spooling Directory Source——HDFS Sink 配置

  • 监听一个指定的目录，只要应用程序向这个指定的目录中添加新的文件，Source 组件就可以获取到该o数据，并解析该文件的内容，然后写入到 Channle。写入完成后，标记该文件已完成或者删除该文件。 其中 Sink：HDFS， Channel：file

    

  1. Property Name       Default      Description
    2. 

  2. channels              –  
    3. 

  3. type                  –          The component type name, needs to be spooldir.
    4. 

  4. spoolDir              –          Spooling Directory Source监听的目录
    5. 

  5. fileSuffix         .COMPLETED    文件内容写入到channel之后，标记该文件
    6. 

  6. deletePolicy       never         文件内容写入到channel之后的删除策略: never or immediate
    7. 

  7. fileHeader         false         Whether to add a header storing the absolute path filename.
    8. 

  8. ignorePattern      ^$           Regular expression specifying which files to ignore (skip)
    9. 

  9. interceptors          –          指定传输中event的head(头信息)，常用timestamp
    10.

  • 配置文件

    

  1. # Name the components on this agent
    2. 

  2. a1.sources = r1
    3. 

  3. a1.sinks = k1
    4. 

  4. a1.channels = c1
    5. 

    6. 

  6. # Describe/configure the source
    7. 

  7. a1.sources.r1.type = spooldir
    8. 

  8. a1.sources.r1.spoolDir = /usr/local/datainput
    9. 

  9. a1.sources.r1.fileHeader = true
    10. 

  10. a1.sources.r1.interceptors = i1
    11. 

  11. a1.sources.r1.interceptors.i1.type = timestamp
    12. 

    13. 

  13. # Describe the sink
    14. 

  14. # Describe the sink
    15. 

  15. a1.sinks.k1.type = hdfs
    16. 

  16. a1.sinks.k1.hdfs.path = hdfs://hadoop238:9000/dataoutput
    17. 

  17. a1.sinks.k1.hdfs.writeFormat = Text
    18. 

  18. a1.sinks.k1.hdfs.fileType = DataStream
    19. 

  19. a1.sinks.k1.hdfs.rollInterval = 10
    20. 

  20. a1.sinks.k1.hdfs.rollSize = 0
    21. 

  21. a1.sinks.k1.hdfs.rollCount = 0
    22. 

  22. a1.sinks.k1.hdfs.filePrefix = %Y-%m-%d-%H-%M-%S
    23. 

  23. a1.sinks.k1.hdfs.useLocalTimeStamp = true
    24. 

    25. 

  25. # Use a channel which buffers events in file
    26. 

  26. a1.channels.c1.type = file
    27. 

  27. a1.channels.c1.checkpointDir = /usr/flume/checkpoint
    28. 

  28. a1.channels.c1.dataDirs = /usr/flume/data
    29. 

    30. 

  30. # Bind the source and sink to the channel
    31. 

  31. a1.sources.r1.channels = c1
    32. 

  32. a1.sinks.k1.channel = c1
    33.

  • 启动 Flume agent a1 服务端

  1. flume-ng  agent -n a1  -c ../conf  -f ../conf/flume-spool-hdfs.conf   -Dflume.root.logger=DEBUG,console

  • 使用 cp 命令向 Spooling Directory 中发送数据

  1. cp datafile  /usr/local/datainput   (注：datafile中的内容为：big data world！)

  • Spooling Directory Source的两个注意事项

    

  1. 1.拷贝到spool目录下的文件不可以再打开编辑
    2. 

  2. 2.不能将具有相同文件名字的文件拷贝到这个目录下
    3.

Exec Source——HDFS Sink 配置

  • 监听一个指定的命令，获取一条命令的结果作为它的数据源 常用的是 tail-F file 指令，只要应用程序向日志(文件)里面写数据，Source 组件就可以获取到日志(文件)中最新的内容 。 其中 Sink：HDFS， Channel：file

  • 配置文件

    

  1. # Name the components on this agent
    2. 

  2. a1.sources = r1
    3. 

  3. a1.sinks = k1
    4. 

  4. a1.channels = c1
    5. 

    6. 

  6. # Describe/configure the source
    7. 

  7. a1.sources.r1.type = exec
    8. 

  8. a1.sources.r1.command = tail -F /usr/local/log.file
    9. 

    10. 

  10. # Describe the sink
    11. 

  11. a1.sinks.k1.type = hdfs
    12. 

  12. a1.sinks.k1.hdfs.path = hdfs://hadoop238:9000/dataoutput
    13. 

  13. a1.sinks.k1.hdfs.writeFormat = Text
    14. 

  14. a1.sinks.k1.hdfs.fileType = DataStream
    15. 

  15. a1.sinks.k1.hdfs.rollInterval = 10
    16. 

  16. a1.sinks.k1.hdfs.rollSize = 0
    17. 

  17. a1.sinks.k1.hdfs.rollCount = 0
    18. 

  18. a1.sinks.k1.hdfs.filePrefix = %Y-%m-%d-%H-%M-%S
    19. 

  19. a1.sinks.k1.hdfs.useLocalTimeStamp = true
    20. 

    21. 

  21. # Use a channel which buffers events in file
    22. 

  22. a1.channels.c1.type = file
    23. 

  23. a1.channels.c1.checkpointDir = /usr/flume/checkpoint
    24. 

  24. a1.channels.c1.dataDirs = /usr/flume/data
    25. 

    26. 

  26. # Bind the source and sink to the channel
    27. 

  27. a1.sources.r1.channels = c1
    28. 

  28. a1.sinks.k1.channel = c1
    29.

  • 启动 Flume agent a1 服务端

  1. flume-ng  agent -n a1  -c ../conf  -f ../conf/flume-exec-hdfs.conf   -Dflume.root.logger=DEBUG,console

  • 使用 echo 命令向 /usr/local/datainput 中发送数据

  1. echo  big data > log.file

Avro Source——HDFS Sink 配置

  • 监听一个指定的 Avro 端口，通过 Avro 端口可以获取到 Avro Client 发送过来的文件 。只要应用程序通过 Avro 端口发送文件，Source 组件就可以获取到该文件中的内容。 其中 Sink：HDFS，Channel：file (注：Avro 和 Thrift 都是一些序列化的网络端口–通过这些网络端口可以接受或者发送信息，Avro 可以发送一个给定的文件给 Flume，是 Agent 间的通信协议)

  • 编写配置文件

    

  1. # Name the components on this agent
    2. 

  2. a1.sources = r1
    3. 

  3. a1.sinks = k1
    4. 

  4. a1.channels = c1
    5. 

    6. 

  6. # Describe/configure the source
    7. 

  7. a1.sources.r1.type = avro
    8. 

  8. a1.sources.r1.bind = 192.168.111.238
    9. 

  9. a1.sources.r1.port = 4141
    10. 

    11. 

  11. # Describe the sink
    12. 

  12. a1.sinks.k1.type = hdfs
    13. 

  13. a1.sinks.k1.hdfs.path = hdfs://hadoop238:9000/dataoutput
    14. 

  14. a1.sinks.k1.hdfs.writeFormat = Text
    15. 

  15. a1.sinks.k1.hdfs.fileType = DataStream
    16. 

  16. a1.sinks.k1.hdfs.rollInterval = 10
    17. 

  17. a1.sinks.k1.hdfs.rollSize = 0
    18. 

  18. a1.sinks.k1.hdfs.rollCount = 0
    19. 

  19. a1.sinks.k1.hdfs.filePrefix = %Y-%m-%d-%H-%M-%S
    20. 

  20. a1.sinks.k1.hdfs.useLocalTimeStamp = true
    21. 

    22. 

  22. # Use a channel which buffers events in file
    23. 

  23. a1.channels.c1.type = file
    24. 

  24. a1.channels.c1.checkpointDir = /usr/flume/checkpoint
    25. 

  25. a1.channels.c1.dataDirs = /usr/flume/data
    26. 

    27. 

  27. # Bind the source and sink to the channel
    28. 

  28. a1.sources.r1.channels = c1
    29. 

  29. a1.sinks.k1.channel = c1
    30.

  • 启动 Flume agent a1 服务端

  1. flume-ng  agent -n a1  -c ../conf  -f ../conf/flume-avro-hdfs.conf   -Dflume.root.logger=DEBUG,console

  • 使用avro-client发送文件

  1. flume-ng avro-client -c  ../conf  -H 192.168.111.238  -p 4141 -F /usr/local/log.file

总结

总结 Exec Source 和 Spooling Directory Source 是两种常用的日志采集的方式，其中 Exec Source 可以对日志的实时采集，但是当 Flume 不运行或者指令执行出错时，Exec Source 将无法收集到日志数据，日志会出现丢失，从而无法保证收集日志的完整性。而 Spooling Directory Source 在对日志的实时采集上有欠缺。


原文作者：HelloTech技术派；如有侵权，请联系管理员删除。

0
0
上一篇： Flume案例——日志分析采集系统
下一篇： Flume核心概念
请登录后发表评论 登录
全部评论
HelloTech技术派

注册时间：2017-12-05

  • 博文量
    181
  • 访问量
    73726

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593