Flume：搭建配置以及 source读取在netcat、http，sink 落实在本地、HDFS

转x33g5p2x  于2021-11-29 转载在 Flume  
字(6.0k)|赞(0)|评价(0)|浏览(743)

1、下载创建并解压Flume

下载地址：http://www.apache.org/dyn/closer.lua/flume/1.7.0/apache-flume-1.7.0-bin.tar.gz

云盘链接：https://pan.baidu.com/s/1dFJ6uhgJcSUGwAHiIOrCDA
提取码：ycjf

下载好，上传到虚拟机节点，进行解压。先创建目录。

  1. # 创建目录
  2. mkdir -p /usr/flume
  3. cd /usr/flume
  4. # 解压到flume目录
  5. tar -zxvf ./apache-flume-1.7.0-bin.tar.gz -C /usr/flume/
  6. # cd /usr/flume/,重命名一下
  7. mv apache-flume-1.7.0-bin/ ./flume-1.7.0

2、配置文件

2.1 配置 flume-env.sh文件

  1. cd /usr/flume/flume-1.7.0/conf
  2. ls
  3. mv flume-env.sh.template flume-env.sh
  4. vi flume-env.sh
  5. ## 添加jdk路径：
  6. export JAVA_HOME=/usr/java/jdk1.8.0_171

2.2 配置Flume环境变量(三个节点)

在 /etc/profile 配置Flume环境变量。

  1. vi /etc/profile
  3. # flume
  4. export FLUME_HOME=/usr/flume/flume-1.7.0
  5. export PATH=$PATH:$FLUME_HOME/bin
  6. export FLUME_CONF_DIR=$FLUME_HOME/conf
  8. source /etc/profile

2.3 配置conf

通过netcat作为source, sink为logger的方式，配置example.conf。

  • agent进行重命名： a1
  • sources： r1
  • sinks: k1
  • channels: c1
  1. # 1、通过netcat作为source, sink为logger的方式
  2. # Name the components on this agent
  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1
  7. # Describe/configure the source
  8. a1.sources.r1.type = netcat
  9. a1.sources.r1.bind = localhost
  10. a1.sources.r1.port = 44444
  12. # Describe the sink
  13. a1.sinks.k1.type = logger
  15. # Use a channel which buffers events in memory
  16. a1.channels.c1.type = memory
  17. a1.channels.c1.capacity = 1000
  18. a1.channels.c1.transactionCapacity = 100
  20. # Bind the source and sink to the channel
  21. a1.sources.r1.channels = c1
  22. a1.sinks.k1.channel = c1

cd /usr/flume/flume-1.7.0，在此目录下启动Flume。

  1. /usr/flume/flume-1.7.0/bin/flume-ng agent --conf conf --conf-file /usr/flume/flume-1.7.0/conf/example.conf -name a1 -Dflume.root.logger=INFO,console

再启动master节点，输入 telent localhost 44444

输入数据之后，另一个master就会接收到数据并提示！

当关闭flume，telnet也会自动断开。

2.4 分发文件

  1. scp -r /usr/flume/ root@slave1:/usr/
  2. scp -r /usr/flume/ root@slave2:/usr/

3、Flume的conf 多种部署

3.1 需求：显示结果是过滤数据

通过netcat作为source, sink为logger的方式，现在我之关心字母，过滤掉数字。

重新编辑 example.conf，添加正则表达。

  1. # 添加source定义正则匹配规则
  2. a1.sources.r1.interceptors = i1  
  3. a1.sources.r1.interceptors.i1.type =regex_filter  
  4. a1.sources.r1.interceptors.i1.regex =^[0-9]*$  
  5. a1.sources.r1.interceptors.i1.excludeEvents =true

保存退出，再启动flume。

  1. ./bin/flume-ng agent --conf conf --conf-file ./conf/example.conf -name a1 -Dflume.root.logger=INFO,console

telnet localhost 44444，输入数据 12，11，bad，hello java。

可以发现接收的数据是过滤了数字！！。

3.2 需求：通过netcat作为source, sink写到hdfs

编辑：vi examp.conf

  1. # Name the components on this agent 针对agent重命名为a1
  2. a1.sources = r1  # source别名为 r1
  3. a1.sinks = k1    # sinks别名为 k1
  4. a1.channels = c1 # channel别名为 c1
  6. # Describe/configure the source
  7. a1.sources.r1.type = netcat
  8. a1.sources.r1.bind = localhost
  9. a1.sources.r1.port = 44444
  11. # source定义正则匹配规则
  12. a1.sources.r1.interceptors = i1  
  13. a1.sources.r1.interceptors.i1.type =regex_filter  
  14. a1.sources.r1.interceptors.i1.regex =^[0-9]*$  
  15. a1.sources.r1.interceptors.i1.excludeEvents =true
  17. # Describe the sink
  18. # a1.sinks.k1.type = logger
  19. a1.channels = c1
  20. a1.sinks = k1
  21. # sink为hdfs
  22. a1.sinks.k1.type = hdfs
  23. # 数据存储到hdfs的文件路径
  24. a1.sinks.k1.hdfs.path = hdfs:/flume
  25. # 表示最终的文件前缀
  26. a1.sinks.k1.hdfs.filePrefix = events
  27. # 表示到了需要触发的时间时，是否要更新文件夹，true:表示是
  28. a1.sinks.k1.hdfs.round = true
  29. a1.sinks.k1.hdfs.roundValue = 10
  30. # 表示切换时间的单位是分钟
  31. a1.sinks.k1.hdfs.roundUnit = minute
  32. # 表示过了一分钟生成一个文件
  33. a1.sinks.k1.hdfs.roundInterval = 60 
  34. a1.sinks.k1.hdfs.fileType = DataStream
  36. # Use a channel which buffers events in 
  37. # memory channel配置
  38. a1.channels.c1.type = memory
  39. a1.channels.c1.capacity = 1000
  40. a1.channels.c1.transactionCapacity = 100
  42. # Bind the source and sink to the channel
  43. # 将source channel sink进行串联起来
  44. a1.sources.r1.channels = c1
  45. a1.sinks.k1.channel = c1

启动flume，在example.conf配置中，我把每隔一分钟输入的数据生成一个文件到HDFS的flume目录下，文件前缀名有event。

  1. ./bin/flume-ng agent --conf conf --conf-file ./conf/example.conf -name a1 -Dflume.root.logger=INFO,console

可以看到一分钟内它还是个临时文件tmp，等一分钟后就变成一个文件在HDFS了。
查看HDFS是否有flume保存的文件。

  1. # 如果事先HDFS已经创建了flume，要先删除
  2. hadoop fs -ls /flume
  3. hadoop fs -cat /flume/events.1638113932881

可以发现，我们是把sink的数据存到HDFS了。
我们有发现一个问题：就是假如一分钟内输入的flume数据过小，那就极大浪费存储空间，我们需要限定一下，设置flume防止小文件。

  1. # 1、限定一个文件的文件数据大小
  2. a1.sinks.k1.hdfs.rollSize = 200*1024*1024
  3. # 2、限定文件可以存储多少个event
  4. a1.sinks.k1.hdfs.rollCount = 10000

3.3 通过HTTP作为source, sink写到logger

创建http.conf， 在/usr/flume/flume-1.7.0/conf，vi http.conf。

  1. # HTTP.conf
  2. # Name the components on this agent
  3. a1.sources = r1
  4. a1.sinks = k1
  5. a1.channels = c1
  7. # Describe/configure the source
  8. a1.sources.r1.type = org.apache.flume.source.http.HTTPSource
  9. a1.sources.r1.bind = master
  10. a1.sources.r1.port = 50020
  11. #a1.sources.r1.fileHeader = true
  13. # Describe the sink
  14. a1.sinks.k1.type = logger
  16. # Use a channel which buffers events in memory
  17. a1.channels.c1.type = memory
  18. a1.channels.c1.capacity = 1000
  19. a1.channels.c1.transactionCapacity = 100
  21. # Bind the source and sink to the channel
  22. a1.sources.r1.channels = c1
  23. a1.sinks.k1.channel = c1

配置完成，启动Flume。在 /usr/flume/flume-1.7.0 目录。

  1. ./bin/flume-ng agent --conf conf --conf-file ./conf/http.conf -name a1 -Dflume.root.logger=INFO,console

输入数据：

  1. curl -X POST -d '[{"headers" : {"timestamp" : "434324343","host" : "random_host.example.com"},"body" : "random_body"},{"headers" : {"namenode" : "namenode.example.com","datanode" : "random_datanode.example.com"},"body" : "badou,badou"}]' master:50020

这个命令也可以在其他节点执行。http是允许在不同结点发送消息给source的。

3.4 多节点进行串联

在slave1 编辑文件pull.conf
vi pull.conf

  1. #Name the components on this agent
  2. a2.sources= r1
  3. a2.sinks= k1
  4. a2.channels= c1
  6. #Describe/configure the source
  7. a2.sources.r1.type= avro
  8. a2.sources.r1.channels= c1
  9. a2.sources.r1.bind= slave1
  10. a2.sources.r1.port= 44444
  12. #Describe the sink
  13. a2.sinks.k1.type= logger
  14. a2.sinks.k1.channel = c1
  16. #Use a channel which buffers events in memory
  17. a2.channels.c1.type= memory
  18. a2.channels.c1.keep-alive= 10
  19. a2.channels.c1.capacity= 100000
  20. a2.channels.c1.transactionCapacity= 100000

在master节点编辑 vi push.conf

  1. #Name the components on this agent
  2. a1.sources= r1
  3. a1.sinks= k1
  4. a1.channels= c1
  6. #Describe/configure the source
  7. a1.sources.r1.type= netcat
  8. a1.sources.r1.bind= localhost
  9. a1.sources.r1.port = 44444
  10. a1.sources.r1.channels= c1
  12. #Use a channel which buffers events in memory
  13. a1.channels.c1.type= memory
  14. a1.channels.c1.keep-alive= 10
  15. a1.channels.c1.capacity= 100000
  16. a1.channels.c1.transactionCapacity= 100000
  18. #Describe/configure the source
  19. a1.sinks.k1.type= avro
  20. a1.sinks.k1.channel= c1
  21. a1.sinks.k1.hostname= slave1
  22. a1.sinks.k1.port= 44444

  1. # 1、slave1启动:
  2. ./bin/flume-ng agent -c conf -f conf/pull.conf -n a2 -Dflume.root.logger=INFO,console
  4. # 2、master启动:
  5. ./bin/flume-ng agent -c conf -f conf/push.conf -n a1 -Dflume.root.logger=INFO,console
  7. 在master上执行 telnet localhost 44444

可以发现我们在master输入的数据，在salve1 进行显示。

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/weixin_44775255/article/details/121599487

内容来源于网络，如有侵权，请联系作者删除！

hadoopflume大数据大数据 linux

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com