第一章.Flume概述

1.定义

Flume是Cloudera提供的一个高可用的,高可靠的,分布式的海量日志采集,聚合和传输的系统,Flume基于流式架构,灵活简单

$01[Flume_概述_入门] - 图1

2.组成架构

$01[Flume_概述_入门] - 图2

  1. # Agent
  2. - Agent是一个JVM进程.它以事件的形式将数据从源头送至目的
  3. - Agent主要有我个部分组成: Source, Channel,Sink
  4. # Source
  5. - Source是负责接收数据到Flume Agent 的组件,Source组件可以处理各种类型,各种格式的日志数据,包括avro,thrift,exec,jms,spooling directory,netcat,taildir,sequence generator,syslog,http,legacy
  6. # Sink
  7. - Sink不断的轮询Channel中的事件且批量的移除它们,并将这些事件批量写入到存储或索引系统,或者被发送到另一个Flume Agent
  8. - Sink组件目的地包括HDFS,logger,avro,thrift,ipc,file,Hbase,solr,自定义
  9. # Channel
  10. - Channel是位于source与sink之间的缓冲区,因此,Channel允许Source和Sink运作在不同的频率上,Channel是线程安全的,可以同时处理几个source的写入操作和几个Sink的读取操作
  11. - Flume自带两种Channel: Memory Channel 和File Channel
  12. - Memory Channel是内存中的队列,Memory Channel在不需要关心数据丢失的情景下适用,如果需要关心数据丢失,那么Memory Channel就不应该使用,因为程序死亡,机器宕机或者重启就会导致数据丢失
  13. - File Channel将所有事件写到磁盘,因此在程序宕机或关闭的情况下不会丢失数据
  14. # Event
  15. - 传输单元,Flume数据传输的基本单元,以Event的形式经数据从源头送至目的地,Event由Header和Body两部分组成,Header用来存放该Event的一些属性,为K-V结构,Body用来存放该条数据,形式为字节数组
Header Body
K-V byte array

第二章.Flume入门

1. Flume安装

安装部署

  1. 将apache-flume-1.9.0-bin.tar.gz上传到linux的/opt/software目录下
  2. 解压apache-flume-1.9.0-bin.tar.gz到/opt/module/目录下
  1. tar -zxvf /opt/software/apache-flume-1.9.0-bin.tar.gz -C /opt/module/
  1. 修改apache-flume-1.9.0-bin的名称为flume-1.9.0
  1. mv /opt/module/apache-flume-1.9.0-bin /opt/module/flume-1.9.0
  1. 配置环境变量
  1. # 打开配置文件
  2. vim /etc/profile.d/my_env.sh
  3. # 添加如下内容
  4. export FLUME_HOME=/opt/module/flume-1.9.0
  5. export PATH=$PATH:$FLUME_HOME/bin
  6. # source 配置文件
  7. source /etc/profile.d/my_env.sh
  1. 将lib文件夹下的guava-11.0.2.jar删除以兼容Hadoop 3.1.3
  1. rm -rf /opt/module/flume-1.9.0/lib/guava-11.0.2.jar

2.Flume入门案例

案例一:监控端口数据官方案例

需求 : 使用Flume监听一个端口,收集该端口数据,并打印到控制台

  1. 安装netcat工具
  1. sudo yum install -y nc
  1. 在flume-1.9.0文件夹下创建目录object/simpleCase/config

$01[Flume_概述_入门] - 图3

  1. 在config目录下创建配置文件flume_1_netcat_logger.conf,并添加如下内容
  1. # 案例1 端口----->logger
  3. # agent
  4. a1.sources = r1
  5. a1.sinks = k1
  6. a1.channels = c1
  8. # source
  9. a1.sources.r1.type = netcat
  10. a1.sources.r1.bind = hadoop102
  11. a1.sources.r1.port = 6666
  13. # sink
  14. a1.sinks.k1.type = logger
  16. # channel
  17. a1.channels.c1.type = memory
  18. a1.channels.c1.capacity = 1000
  19. a1.channels.c1.transactionCapacity = 100
  21. # bind
  22. a1.sources.r1.channels = c1
  23. a1.sinks.k1.channel = c1
  1. 开启Flume监听端口
  1. flume-ng agent --name a1 --conf /opt/module/flume-1.9.0/conf/ --conf-file flume_1_netcat_logger.conf -Dflume.root.logger=INFO,console
  1. 使用netcat工具向本机的6666端口发送内容
  1. nc hadoop102 6666
  2. hello
  3. jcsune
  1. 在flume 监听页面观察接收数据情况

$01[Flume_概述_入门] - 图4

案例二:实时监控单个追加文件

需求:实时监控Hive日志,并上传到HDFS中

需求分析

$01[Flume_概述_入门] - 图5

实现步骤

  1. 确保Hadoop和java环境配置正确
  2. 创建flume_2_exec_hdfs.conf文件,并添加如下内容
  1. # 案例2  实时监控单个追加文件
  3. # agent
  4. a2.sources=r1
  5. a2.sinks=k1
  6. a2.channels=c1
  8. # source
  9. a2.sources.r1.type = exec
  10. a2.sources.r1.command = tail -F /opt/module/flume-1.9.0/object/simpleCase/data/case2.log
  11. a2.sources.r1.shell = /bin/bash -c
  13. # sink
  14. a2.sinks.k1.type = hdfs
  15. a2.sinks.k1.hdfs.path = hdfs://hadoop102:9820/flume-1.9.0/simpleCase/2/%Y%m%d/%H
  16. #上传文件的前缀
  17. a2.sinks.k1.hdfs.filePrefix = logs-
  18. #是否按照时间滚动文件夹
  19. a2.sinks.k1.hdfs.round = true
  20. #多少时间单位创建一个新的文件夹
  21. a2.sinks.k1.hdfs.roundValue = 1
  22. #重新定义时间单位
  23. a2.sinks.k1.hdfs.roundUnit = hour
  24. #是否使用本地时间戳
  25. a2.sinks.k1.hdfs.useLocalTimeStamp = true
  26. #积攒多少个Event才flush到HDFS一次
  27. a2.sinks.k1.hdfs.batchSize = 100
  28. #设置文件类型，可支持压缩
  29. a2.sinks.k1.hdfs.fileType = DataStream
  30. #多久生成一个新的文件
  31. a2.sinks.k1.hdfs.rollInterval = 60
  32. #设置每个文件的滚动大小
  33. a2.sinks.k1.hdfs.rollSize = 134217700
  34. #文件的滚动与Event数量无关
  35. a2.sinks.k1.hdfs.rollCount = 0
  38. # channel
  39. a2.channels.c1.type = memory
  40. a2.channels.c1.capacity = 1000
  41. a2.channels.c1.transactionCapacity = 100
  43. # bind
  44. a2.sources.r1.channels=c1
  45. a2.sinks.k1.channel=c1
  1. 运行Flume
  1. flume-ng agent --name a2 --conf /opt/module/flume-1.9.0/conf/ --conf-file flume_2_exec_hdfs.conf -Dflume.root.logger=INFO,console
  1. 开启Hadoop和Hive并操作Hive产生日志
  1. mycluster.sh start #开启Hadoop
  2. hiveservice.sh start #开启Hive
  1. 测试(/opt/module/flume-1.9.0/object/simpleCase/data)
  1. echo 1 >> case2.log
  2. echo 2 >> case2.log
  3. echo 3 >> case2.log
  4. echo 4 >> case2.log
  5. echo 5 >> case2.log
  1. 浏览器查看

$01[Flume_概述_入门] - 图6

案例三:实时监控目录下多个新文件

案例需求: 使用Flume监听整个目录的文件,并上传至HDFS

$01[Flume_概述_入门] - 图7

实现步骤:

  1. 创建配置文件flume_3_spoolDir_hdfs.conf,并添加如下内容
  1. # 案例3 实时监控目录下多个文件
  3. # agent
  4. a3.sources=r1
  5. a3.sinks=k1
  6. a3.channels=c1
  8. # source
  9. a3.sources.r1.type = spooldir
  10. a3.sources.r1.spoolDir = /opt/module/flume-1.9.0/object/simpleCase/data/
  11. a3.sources.r1.fileSuffix = .finish
  12. a3.sources.r1.fileHeader = true
  13. #忽略所有以.tmp结尾的文件，不上传
  14. a3.sources.r1.ignorePattern = ([^ ]*\.tmp)
  17. # sink
  18. a3.sinks.k1.type = hdfs
  19. a3.sinks.k1.hdfs.path = hdfs://hadoop102:9820/flume-1.9.0/simpleCase/3/%Y%m%d/%H
  20. #上传文件的前缀
  21. a3.sinks.k1.hdfs.filePrefix = logs-
  22. #是否按照时间滚动文件夹
  23. a3.sinks.k1.hdfs.round = true
  24. #多少时间单位创建一个新的文件夹
  25. a3.sinks.k1.hdfs.roundValue = 1
  26. #重新定义时间单位
  27. a3.sinks.k1.hdfs.roundUnit = hour
  28. #是否使用本地时间戳
  29. a3.sinks.k1.hdfs.useLocalTimeStamp = true
  30. #积攒多少个Event才flush到HDFS一次
  31. a3.sinks.k1.hdfs.batchSize = 100
  32. #设置文件类型，可支持压缩
  33. a3.sinks.k1.hdfs.fileType = DataStream
  34. #多久生成一个新的文件
  35. a3.sinks.k1.hdfs.rollInterval = 60
  36. #设置每个文件的滚动大小
  37. a3.sinks.k1.hdfs.rollSize = 134217702
  38. #文件的滚动与Event数量无关
  39. a3.sinks.k1.hdfs.rollCount = 0
  42. # channel
  43. a3.channels.c1.type = memory
  44. a3.channels.c1.capacity = 1000
  45. a3.channels.c1.transactionCapacity = 100
  47. # bind
  48. a3.sources.r1.channels=c1
  49. a3.sinks.k1.channel=c1
  1. 启动监控文件夹命令
  1. flume-ng agent --name a3 --conf /opt/module/flume-1.9.0/conf/ --conf-file flume_3_spoolDir_hdfs.conf -Dflume.root.logger=INFO,console
  1. 在/opt/module/flume-1.9.0/object/simpleCase下新建test文件夹并添加文件

$01[Flume_概述_入门] - 图8

  1. 浏览器端查看

$01[Flume_概述_入门] - 图9

案例四:实时监控目录下多个追加文件

Exec source适用于监控一个实时追加的文件，不能实现断点续传；Spooldir Source适合用于同步新文件，但不适合对实时追加日志的文件进行监听并同步；而Taildir Source适合用于监听多个实时追加的文件，并且能够实现断点续传。

案例需求:使用Flume监听整个目录的实时追加文件,并上传到HDFS

$01[Flume_概述_入门] - 图10

实现步骤:

  1. 创建配置文件flume_4_tailDir_hdfs.conf ,并添加如下内容
  1. # 案例四 实时监控目录下多个追加文件
  3. # agent
  4. a4.sources=r1
  5. a4.sinks=k1
  6. a4.channels=c1
  8. # source
  9. a4.sources.r1.type = TAILDIR
  10. a4.sources.r1.positionFile = /opt/module/flume-1.9.0/tail_dir.json
  11. a4.sources.r1.filegroups = f1 f2
  12. a4.sources.r1.filegroups.f1 = /opt/module/flume-1.9.0/object/simpleCase/file/.*file.*
  13. a4.sources.r1.filegroups.f2 = /opt/module/flume-1.9.0/object/simpleCase/log/.*log.*
  16. # sink
  17. a4.sinks.k1.type = hdfs
  18. a4.sinks.k1.hdfs.path = hdfs://hadoop102:9820/flume-1.9.0/simpleCase/4/%Y%m%d/%H
  19. #上传文件的前缀
  20. a4.sinks.k1.hdfs.filePrefix = case4-
  21. #是否按照时间滚动文件夹
  22. a4.sinks.k1.hdfs.round = true
  23. #多少时间单位创建一个新的文件夹
  24. a4.sinks.k1.hdfs.roundValue = 1
  25. #重新定义时间单位
  26. a4.sinks.k1.hdfs.roundUnit = hour
  27. #是否使用本地时间戳
  28. a4.sinks.k1.hdfs.useLocalTimeStamp = true
  29. #积攒多少个Event才flush到HDFS一次
  30. a4.sinks.k1.hdfs.batchSize = 100
  31. #设置文件类型，可支持压缩
  32. a4.sinks.k1.hdfs.fileType = DataStream
  33. #多久生成一个新的文件
  34. a4.sinks.k1.hdfs.rollInterval = 60
  35. #设置每个文件的滚动大小
  36. a4.sinks.k1.hdfs.rollSize = 134217700
  37. #文件的滚动与Event数量无关
  38. a4.sinks.k1.hdfs.rollCount = 0
  41. # channel
  42. a4.channels.c1.type = memory
  43. a4.channels.c1.capacity = 1000
  44. a4.channels.c1.transactionCapacity = 100
  46. # bind
  47. a4.sources.r1.channels=c1
  48. a4.sinks.k1.channel=c1
  1. 启动监控文件夹命令
  1. flume-ng agent --name a4 --conf /opt/module/flume-1.9.0/conf/ --conf-file flume_4_tailDir_hdfs.conf -Dflume.root.logger=INFO,console
  1. 分别创建好测试文件

$01[Flume_概述_入门] - 图11

$01[Flume_概述_入门] - 图12

  1. 浏览器查看

$01[Flume_概述_入门] - 图13