• 尚硅谷Flume（仅有基础）

    q

    1 概述

    1.1 定义

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

            Flume最主要的作用就是，实时读取服务器本地磁盘的数据，将数据写入到HDFS。

    1.2 架构

    1.2.1 Agent 

            Agent是一个JVM进程，它以事件的形式将数据从源头送至目的。 
            Agent主要有3个部分组成，Source、Channel、Sink。 

    1.2.2 Source   

            Source是负责接收数据到Flume Agent的组件。Source组件可以处理各种类型、各种格式的日志数据，包括avro、thrift、exec、jms、spooling directory、netcat、taildir、sequence generator、syslog、http、legacy。 

    1.2.3 Sink 

            Sink 不断地轮询 Channel 中的事件且批量地移除它们，并将这些事件批量写入到存储或索引系统、或者被发送到另一个 Flume Agent。 

    Sink 组件目的地包括 hdfs、logger、avro、thrift、ipc、file、HBase、solr、自定义。 

    1.2.4 Channel 

            Channel是位于Source 和Sink 之间的缓冲区。因此，Channel允许 Source和Sink 运作在不同的速率上。Channel 是线程安全的，可以同时处理几个 Source 的写入操作和几个Sink的读取操作。 
            Flume自带两种Channel：Memory Channel和 File Channel。 

            Memory Channel是内存中的队列。Memory Channel在不需要关心数据丢失的情景下适用。如果需要关心数据丢失，那么 Memory Channel就不应该使用，因为程序死亡、机器宕机或者重启都会导致数据丢失。 

            File Channel 将所有事件写到磁盘。因此在程序关闭或机器宕机的情况下不会丢失数据。 

    1.2.5 Event 

            传输单元，Flume 数据传输的基本单元，以 Event 的形式将数据从源头送至目的地。

            Event由Header 和Body 两部分组成，Header用来存放该 event的一些属性，为K-V 结构，Body用来存放该条数据，形式为字节数组。

    2 Flume基本操作

    2.1 安装部署

    http://www.apache.org/dyn/closer.lua/flume/1.9.0/apache-flume-1.9.0-tar.gz

    直接解压

    将lib文件夹下的guava-11.0.2.jar删除以兼容 Hadoop 3.1.3

    2.2 案例

    2.2.1 监控端口数据官方案例

    Flume 1.9.0 User Guide — Apache Flume

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

    （1）安装netcat工具 
    [atguigu@hadoop102 software]$ sudo yum install -y nc 
    （2）判断4444端口是否被占用 
    [atguigu@hadoop102 flume-telnet]$ sudo netstat -nlp | grep 4444
    （3）创建Flume Agent配置文件flume-netcat-logger.conf 
    （4）在flume目录下创建 job文件夹并进入job文件夹。 
    [atguigu@hadoop102 flume]$ mkdir job 
    [atguigu@hadoop102 flume]$ cd job/ 
    （5）在job文件夹下创建 Flume Agent配置文件flume-netcat-logger.conf。 
    [atguigu@hadoop102 job]$ vim flume-netcat-logger.conf 
    （6）在flume-netcat-logger.conf文件中添加如下内容

    1. # name the components on this agent 
    2. a1.sources = r1
    3. a1.sinks = k1
    4. a1.channels = c1
    5.  
    6. # Describe/configure the source 
    7. a1.sources.r1.type = netcat
    8. a1.sources.r1.bind = localhost
    9. a1.sources.r1.port = 4444
    10.  
    11. # Describe the sink
    12. a1.sinks.k1.type = logger
    13.  
    14. # Use a channel which buffers events in memory 
    15. a1.channels.c1.type = memory
    16. a1.channels.c1.capacity = 1000
    17. a1.channels.c1.transactionCapacity = 100
    18.  
    19. # Bind the source and sink to the channel 
    20. a1.sources.r1.channels = c1
    21. a1.sinks.k1.channel = c1

    原神启动  

    1. bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume-netcat-logger.conf -Dflume.root.logger=INFO,console
    2. or
    3. bin/flume-ng agent -c conf/ -n a1 -f job/flume-netcat-logger.conf -flume.root.logger=INFO,console 
    4.  
    5.  --conf/-c：表示配置文件存储在 conf/目录 
    6.  --name/-n：表示给 agent 起名为 a1 
    7.  --conf-file/-f：flume 本次启动读取的配置文件是在 job 文件夹下的 flume-telnet.conf文件。 
    8.  -Dflume.root.logger=INFO,console ：-D 表示 flume 运行时动态修改 flume.root.logge参数属性值，并将控制台日志打印级别设置为 INFO 级别。日志级别包括:log、info、warn、error。

    另一台nc启动

     nc localhost 4444

    然后发消息

    2023-10-25 14:03:53,633 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:95)] Event: { headers:{} body: 30   

    2.2.2 实时监控单个追加文件

    实时监控 Hive 日志，并上传到HDFS中 

    开启hadoop集群

    start-all.sh

    开启hive

    /export/servers/hive/bin/hive --service metastore & nohup /export/servers/hive/bin/hive

    vim flume-file-hdfs.conf

    1. # Name the components on this agent 
    2. a2.sources = r2
    3. a2.sinks = k2
    4. a2.channels = c2
    5.  
    6. # Describe/configure the source 
    7. a2.sources.r2.type = exec
    8. a2.sources.r2.command = tail -F /export/server/hive/logs/hive.log（要监控拉取的文件）
    9.  
    10. # Describe the sink 
    11. a2.sinks.k2.type = hdfs
    12. a2.sinks.k2.hdfs.path = hdfs://hadoop1:8020/flume/%Y%m%d/%H（这里的端口要和hadoop配置里hdfs的一样！！！！！！！！！）
    13. #上传文件的前缀 
    14. a2.sinks.k2.hdfs.filePrefix = logs-
    15. #是否按照时间滚动文件夹 
    16. a2.sinks.k2.hdfs.round = true
    17. #多少时间单位创建一个新的文件夹 
    18. a2.sinks.k2.hdfs.roundValue = 1
    19. #重新定义时间单位 
    20. a2.sinks.k2.hdfs.roundUnit = hour
    21. #是否使用本地时间戳 
    22. a2.sinks.k2.hdfs.useLocalTimeStamp = true
    23. #积攒多少个Event 才flush 到HDFS一次 
    24. a2.sinks.k2.hdfs.batchSize = 100
    25. #设置文件类型，可支持压缩
    26. a2.sinks.k2.hdfs.fileType = DataStream
    27. #多久生成一个新的文件
    28. a2.sinks.k2.hdfs.rollInterval = 60
    29. #设置每个文件的滚动大小
    30. a2.sinks.k2.hdfs.rollSize = 134217700
    31. #文件的滚动与Event数量无关
    32. a2.sinks.k2.hdfs.rollCount = 0
    33.  
    34. # Use a channel which buffers events in memory 
    35. a2.channels.c2.type = memory
    36. a2.channels.c2.capacity = 1000
    37. a2.channels.c2.transactionCapacity = 100
    38.  
    39. # Bind the source and sink to the channel 
    40. a2.sources.r2.channels = c2
    41. a2.sinks.k2.channel = c2

      启动

    bin/flume-ng agent -n a2 -c conf -f job/flume-file-hdfs.conf

    ctrl+Z退出

    在HDFS上查看文件。

    2.2.3 实时监控目录下多个新文件 

    使用Flume监听整个目录的文件，并上传至 HDFS 

             在使用 Spooling Directory Source 时，不要在监控目录中创建并持续修改文件；上传完成的文件会以.COMPLETED结尾；被监控文件夹每 500毫秒扫描一次文件变动。 

    1. bin/flume-ng agent --conf conf/ --name a3 --conf-file job/flume-dir-hdfs.conf
    2.  
    3.  
    4. a3.sources = r3
    5. a3.sinks = k3
    6. a3.channels = c3
    7.  
    8. # Describe/configure the source
    9. a3.sources.r3.type = spooldir
    10. a3.sources.r3.spoolDir = /export/server/flume/upload
    11. a3.sources.r3.fileSuffix = .COMPLETED
    12. a3.sources.r3.fileHeader = true
    13. #忽略所有以.tmp结尾的文件，不上传 
    14. a3.sources.r3.ignorePattern = ([^ ]*\.tmp)
    15.  
    16. # Describe the sink 
    17. a3.sinks.k3.type = hdfs
    18. a3.sinks.k3.hdfs.path = hdfs://hadoop1:8020/flume/upload/%Y%m%d/%H
    19. #上传文件的前缀:
    20. a3.sinks.k3.hdfs.filePrefix = upload-
    21. #是否按照时间滚动文件夹 
    22. a3.sinks.k3.hdfs.round = true
    23. #多少时间单位创建一个新的文件夹 
    24. a3.sinks.k3.hdfs.roundValue = 1
    25. #重新定义时间单位
    26. a3.sinks.k3.hdfs.roundUnit = hour
    27. #是否使用本地时间戳
    28. a3.sinks.k3.hdfs.useLocalTimeStamp = true
    29. #积攒多少个Event 才flush 到HDFS一次
    30. a3.sinks.k3.hdfs.batchSize = 100
    31. #设置文件类型，可支持压缩 
    32. a3.sinks.k3.hdfs.fileType = DataStream
    33. #多久生成一个新的文件 
    34. a3.sinks.k3.hdfs.rollInterval = 60
    35. #设置每个文件的滚动大小大概是 128M
    36. a3.sinks.k3.hdfs.rollSize = 134217700
    37. #文件的滚动与Event数量无关
    38. a3.sinks.k3.hdfs.rollCount = 0
    39.  
    40. # Use a channel which buffers events in memory 
    41. a3.channels.c3.type = memory
    42. a3.channels.c3.capacity = 1000
    43. a3.channels.c3.transactionCapacity = 100
    44.  
    45. # Bind the source and sink to the channel 
    46. a3.sources.r3.channels = c3
    47. a3.sinks.k3.channel = c3

    在/opt/module/flume 目录下创建upload文件夹 

    向 upload文件夹中添加文件 

    2.2.4 实时监控目录下的多个追加文件 

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

    1. a3.sources = r3 
    2. a3.sinks = k3 
    3. a3.channels = c3 
    4.  
    5. # Describe/configure the source 
    6. a3.sources.r3.type = TAILDIR 
    7. a3.sources.r3.positionFile = /opt/module/flume/tail_dir.json 
    8. a3.sources.r3.filegroups = f1 f2 
    9. a3.sources.r3.filegroups.f1 = /opt/module/flume/files/.*file.* 
    10. a3.sources.r3.filegroups.f2 = /opt/module/flume/files2/.*log.* 
    11.  
    12. # Describe the sink 
    13. a3.sinks.k3.type = hdfs 
    14. a3.sinks.k3.hdfs.path = 
    15. hdfs://hadoop102:9820/flume/upload2/%Y%m%d/%H 
    16. #上传文件的前缀 
    17. a3.sinks.k3.hdfs.filePrefix = upload-
    18. #是否按照时间滚动文件夹 
    19. a3.sinks.k3.hdfs.round = true 
    20. #多少时间单位创建一个新的文件夹 
    21. a3.sinks.k3.hdfs.roundValue = 1 
    22. #重新定义时间单位 
    23. a3.sinks.k3.hdfs.roundUnit = hour 
    24. #是否使用本地时间戳 
    25. a3.sinks.k3.hdfs.useLocalTimeStamp = true 
    26. #积攒多少个Event 才flush 到HDFS一次 
    27. a3.sinks.k3.hdfs.batchSize = 100 
    28. #设置文件类型，可支持压缩 
    29. a3.sinks.k3.hdfs.fileType = DataStream 
    30. #多久生成一个新的文件 
    31. a3.sinks.k3.hdfs.rollInterval = 60 
    32. #设置每个文件的滚动大小大概是128M 
    33. a3.sinks.k3.hdfs.rollSize = 134217700 
    34. #文件的滚动与Event数量无关 
    35. a3.sinks.k3.hdfs.rollCount = 0 
    36.  
    37. # Use a channel which buffers events in memory 
    38. a3.channels.c3.type = memory 
    39. a3.channels.c3.capacity = 1000 
    40. a3.channels.c3.transactionCapacity = 100 
    41.  
    42. # Bind the source and sink to the channel 
    43. a3.sources.r3.channels = c3 
    44. a3.sinks.k3.channel = c3

    Taildir 说明： 
      Taildir Source维护了一个 json格式的position File，其会定期的往position File中更新每个文件读取到的最新的位置，因此能够实现断点续传。Position File的格式如下

    3 Flume 高级

     

  • 相关阅读:
    微信小程序涉嫌通过中断用户体验、限制用户操作的方式，收集与服务无关的用户个人信息，包括但不限于，手机号、
    来自 AI Secure 实验室的 LLM 安全排行榜简介
    Qt之菜单栏、工具栏、状态栏介绍及工具栏QAction的动态增删显示实现方式
    太厉害了，阿里内网最新发布的【MySQL面试小册】把MySQL架构+索引原理+性能优化都讲清楚了
    力扣第236题 二叉树的最近公共祖先 c++ 递归和回溯 附注释和简短代码
    Django文件配置、request对象、连接MySQL、ORM
    Java“牵手”1688商品列表页数据采集+商品价格数据排序，商品销量排序数据，1688API接口采集方法
    jmeter+ant实现的接口自动化测试
    租车小程序系统开发制作功能介绍
    Android 12 DreamCamera2 默认关闭HDR
  • 原文地址：https://blog.csdn.net/qq_58551342/article/details/134031060