• Flume


    Flume

    1.Flume概述

    1.1 Flume定义

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

    1.2 Flume基础架构

    Flume组成架构如下图所示。

    1.2.1 Agent

    Agent是一个JVM进程,它以事件的形式将数据从源头送至目的。

    Agent主要有3个部分组成,SourceChannelSink

    1.2.2 Source

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

    1.2.3 Sink

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

    Sink组件目的地包括hdfsloggeravro、thrift、ipc、fileHBase、solr、自定义。

    1.2.4 Channel

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

    Flume自带两种Channel:Memory ChannelFile Channel

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

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

    1.2.5 Event

    传输单元,Flume数据传输的基本单元,以Event的形式将数据从源头送至目的地。Event由HeaderBody两部分组成,Header用来存放该event的一些属性,为K-V结构,Body用来存放该条数据,形式为字节数组。

    2. Flume安装部署

    解压,改名

    [root@kb129 install]# tar -xvf ./apache-flume-1.9.0-bin.tar.gz -C ../soft/

    复制一份配置文件,并进行配置

    [root@kb129 conf]# cp flume-env.sh.template flume-env.sh

    [root@kb129 conf]# vim ./flume-env.sh

    22 export JAVA_HOME=/opt/soft/jdk180

    25 export JAVA_OPTS="-Xms2000m -Xmx2000m -Dcom.sun.management.jmxremote"

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

    找到flume下现有的guava jar包并删除

    [root@kb129 lib]# find ./ -name guava*

    ./guava-11.0.2.jar

    [root@kb129 lib]# rm -rf ./guava-11.0.2.jar

    拷贝新hadoop内的guava至flume内

    [root@kb129 lib]# pwd

    /opt/soft/hadoop313/share/hadoop/hdfs/lib

    [root@kb129 lib]# cp ./guava-27.0-jre.jar /opt/soft/flume190/lib/

    安装工具

    [root@kb129 conf]# yum install -y net-tools

    [root@kb129 conf]# yum install -y nc           #安装netcat服务

    [root@kb129 conf]# yum install -y telnet-server   #安装netcat服务

    [root@kb129 conf]# yum install -y telnet.*       #安装netcat客户端

    测试工具

    启动服务端口

    [root@kb129 conf]# nc -lk 7777

    连接服务器

    [root@kb129 conf]# telnet localhost 7777

    查看端口是否占用

    [root@kb129 conf]# netstat -lnp | grep 7777

    tcp        0      0 0.0.0.0:7777            0.0.0.0:*               LISTEN      9264/nc            

    tcp6       0      0 :::7777                 :::*                  LISTEN      9264/nc 

    2.2 Flume入门案例

    2.2.1 监控端口数据官方案例

    1)案例需求:

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

    2)需求分析:本次使用的端口为7777

    3)实现步骤:

    创建myconf2文件夹,写入监控配置文件

    [root@kb129 lib]# cd ../conf/myconf2/

    [root@kb129 myconf2]# vim ./netcat-logger.conf

    1. a1.sources=r1
    2. a1.channels=c1
    3. a1.sinks=k1
    4. # Describe/configure the source
    5. a1.sources.r1.type=netcat
    6. a1.sources.r1.bind=localhost
    7. a1.sources.r1.port=7777
    8. # Use a channel which buffers events in memory
    9. a1.channels.c1.type=memory
    10. # Describe the sink
    11. a1.sinks.k1.type=logger
    12. # Bind the source and sink to the channel
    13. a1.sources.r1.channels=c1
    14. a1.sinks.k1.channel=c1

    启动监控命令(a1,conf目录,conf文件,指定控制台输出info信息)

    [root@kb129 flume190]# ./bin/flume-ng agent --name a1 --conf ./conf/ --conf-file ./conf/myconf2/netcat-logger.conf -Dflume.root.logger=INFO,console

    参数说明:

            --conf/-c:表示配置文件存储在conf/目录

            --name/-n:表示给agent起名为a1

            --conf-file/-f:flume本次启动读取的配置文件是在job文件夹下flume-telnet.conf文件。

            -Dflume.root.logger=INFO,console :-D表示flume运行时动态修改flume.root.logger参数属性值,并将控制台日志打印级别设置为INFO级别。日志级别包括:log、info、warn、error。

    [root@kb129 conf]# telnet localhost 7777

    输入内容,控制台可监控输入在Flume监听页面观察接收数据情况

    2.2.2 实时监控单个追加文件

    1)案例需求:使用Flume监听单个的文件

    2)需求分析:

    3)实现步骤:

    [root@kb129 myconf2]# vim ./ filelogger.conf

    1. a2.sources=r1
    2. a2.channels=c1
    3. a2.sinks=k1
    4. # Describe/configure the source
    5. a2.sources.r1.type=exec
    6. a2.sources.r1.command=tail -f /opt/tmp/flumelog.log
    7. # Use a channel which buffers events in memory
    8. a2.channels.c1.type=memory
    9. a2.channels.c1.capacity=1000
    10. a2.channels.c1.transactionCapacity=100
    11. # Describe the sink
    12. a2.sinks.k1.type=logger
    13. # Bind the source and sink to the channel
    14. a2.sources.r1.channels=c1
    15. a2.sinks.k1.channel=c1

    开启flume监听端口

    [root@kb129 flume190]# ./bin/flume-ng agent -n a2 -c ./conf/ -f ./conf/myconf2/filelogger.conf -Dflume.root.logger=INFO,console

    2.2.3 实时监控单个追加文件

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

    2)需求分析

    3)实现步骤:

    [root@kb129 myconf2]# vim ./file-flume-hdfs.conf

    1. a3.sources=r1
    2. a3.sinks=k1
    3. a3.channels=c1
    4. # Describe/configure the source
    5. a3.sources.r1.type=exec
    6. a3.sources.r1.command=tail -f /opt/tmp/flumelog.log
    7. # Use a channel which buffers events in memory
    8. a3.channels.c1.type=memory
    9. a3.channels.c1.capacity=1000
    10. a3.channels.c1.transactionCapacity=100
    11. # Describe the sink
    12. a3.sinks.k1.type=hdfs
    13. a3.sinks.k1.hdfs.fileType=DataStream
    14. a3.sinks.k1.hdfs.filePrefix=flumetohdfs
    15. a3.sinks.k1.hdfs.fileSuffix=.txt
    16. a3.sinks.k1.hdfs.path=hdfs://kb129:9000/kb23flume/
    17. # Bind the source and sink to the channel
    18. a3.sources.r1.channels=c1
    19. a3.sinks.k1.channel=c1

    启动监控

    [root@kb129 flume190]# ./bin/flume-ng agent -n a3 -c ./conf/ -f ./conf/myconf2/file-flume-hdfs.conf -Dflume.root.logger=INFO,console

    2.2.4 实时监控单个追加文件,多个输出

    1)案例需求:使用Flume监听整个目录的文件,并上传至HDFS和本地logger

    2)实现步骤:

    [root@kb129 myconf2]# vim ./file-flume-hdfslogger.conf

    1. a4.sources=r1
    2. a4.channels=c1 c2
    3. a4.sinks=k1 k2
    4. # Describe/configure the source
    5. a4.sources.r1.type=exec
    6. a4.sources.r1.command=tail -f /opt/tmp/flumelog.log
    7. # Use a channel which buffers events in memory
    8. a4.channels.c1.type=memory
    9. a4.channels.c2.type=memory
    10. a4.channels.c1.capacity=1000
    11. a4.channels.c1.transactionCapacity=100
    12. # Describe the sink
    13. a4.sinks.k1.type=logger
    14. a4.sinks.k2.type=hdfs
    15. a4.sinks.k2.hdfs.fileType=DataStream
    16. a4.sinks.k2.hdfs.filePrefix=flumetohdfs
    17. a4.sinks.k2.hdfs.fileSuffix=.txt
    18. a4.sinks.k2.hdfs.path=hdfs://kb129:9000/kb23flume1/
    19. # Bind the source and sink to the channel
    20. a4.sources.r1.channels=c1 c2
    21. a4.sinks.k1.channel=c1
    22. a4.sinks.k2.channel=c2

    启动flume监控

    [root@kb129 flume190]# ./bin/flume-ng agent -n a4 -c ./conf/ -f ./conf/myconf2/file-flume-hdfslogger.conf -Dflume.root.logger=INFO,console

    追加文件,可以在控制台和hdfs中查看到监控日志

    2.2.5 实时监控端口数据,输出至HDFS

    1)案例需求:使用Flume监听整个目录的文件,并上传至HDFS和本地logger

    2)实现步骤:

    使用Flume监听一个端口,收集该端口数据,并输出到hdfs

    [root@kb129 myconf2]# vim ./demo.conf

    1. a5.sources=r1
    2. a5.sinks=k1
    3. a5.channels=c1
    4. # Describe/configure the source
    5. a5.sources.r1.type=netcat
    6. a5.sources.r1.bind=localhost
    7. a5.sources.r1.port=7777
    8. # Use a channel which buffers events in memory
    9. a5.channels.c1.type=memory
    10. a5.channels.c1.capacity=1000
    11. a5.channels.c1.transactionCapacity=100
    12. # Describe the sink
    13. a5.sinks.k1.type=hdfs
    14. a5.sinks.k1.hdfs.fileType=DataStream
    15. a5.sinks.k1.hdfs.filePrefix=flumetohdfs
    16. a5.sinks.k1.hdfs.fileSuffix=.txt
    17. a5.sinks.k1.hdfs.path=hdfs://kb129:9000/kb23flume2/
    18. # Bind the source and sink to the channel
    19. a5.sources.r1.channels=c1
    20. a5.sinks.k1.channel=c1

    启动监听

    [root@kb129 flume190]# ./bin/flume-ng agent -n a5 -c ./conf/ -f ./conf/myconf2/demo.conf -Dflume.root.logger=INFO,console

    开启通信端口,发送数据,在hdfs查看

    2.2.6 实时监控目录下指定文件(spooldir、正则、断点续传、sink至kafka)

    Exec source适用于监控一个实时追加的文件,不能实现断点续传;Spooldir Source适合用于同步新文件,但不适合对实时追加日志的文件进行监听并同步;而Taildir Source适合用于监听多个实时追加的文件,并且能够实现断点续传。总结:exec source从外部命令的输出中读取数据,spooldir source适用于处理已存在的文件,而taildir source适用于实时产生的日志文件。

    1)读取文件,管道类型memory

    [root@kb129 myconf2]# vim events-flume-logger.conf

    1. events.sources=eventsSource
    2. events.channels=eventsChannel
    3. events.sinks=eventsSink
    4. # Describe/configure the source
    5. events.sources.eventsSource.type=spooldir
    6. events.sources.eventsSource.spoolDir=/opt/kb23/flumelogfile/events
    7. #反序列化类型为LINE
    8. events.sources.eventsSource.deserializer=LINE
    9. #反序列化 每行的最大长度
    10. events.sources.eventsSource.deserializer.maxLineLength=32000
    11. events.sources.eventsSource.includePattern=events.csv
    12. # Use a channel which buffers events in memory
    13. events.channels.eventsChannel.type=memory
    14. # Describe the sink
    15. events.sinks.eventsSink.type=logger
    16. # Bind the source and sink to the channel
    17. events.sources.eventsSource.channels= eventsChannel
    18. events.sinks.eventsSink.channel= eventsChannel

    (2)读取文件,加正则过滤,管道类型file

    1. 文件管道(file):文件管道将事件数据写入磁盘上的文件中。这种管道适用于需要持久化存储数据的场景。文件管道可以在系统重启后继续读取未处理的数据,因为数据被写入了磁盘。但是,由于写入和读取都需要磁盘操作,所以文件管道的性能可能相对较低。

    2. 内存管道(memory):内存管道将事件数据保存在内存中。这种管道适用于对性能要求较高的场景,因为内存操作比磁盘操作更快。内存管道不会将数据持久化到磁盘,所以在系统重启后,未处理的数据将会丢失。因此,内存管道适用于对数据可靠性要求不高的场景。

    [root@kb129 myconf2]# vim events-flume-logger.conf

    1. events.sources=eventsSource
    2. events.channels=eventsChannel
    3. events.sinks=eventsSink
    4. # Describe/configure the source
    5. events.sources.eventsSource.type=spooldir
    6. events.sources.eventsSource.spoolDir=/opt/kb23/flumelogfile/events
    7. #反序列化类型为LINE
    8. events.sources.eventsSource.deserializer=LINE
    9. #反序列化 每行的最大长度
    10. events.sources.eventsSource.deserializer.maxLineLength=32000
    11. events.sources.eventsSource.includePattern=events_[0-9]{4}-[0-9]{2}-[0-9]{2}.csv
    12. #正则过滤器
    13. events.sources.eventsSource.interceptors=head_filter
    14. events.sources.eventsSource.interceptors.head_filter.type=regex_filter
    15. events.sources.eventsSource.interceptors.head_filter.regex=^event_id*
    16. events.sources.eventsSource.interceptors.head_filter.excludeEvents=true
    17. # Use a channel which buffers events in memory
    18. events.channels.eventsChannel.type=file
    19. events.channels.eventsChannel.checkpointDir=/opt/kb23/checkpoint/events
    20. events.channels.eventsChannel.dataDirs=/opt/kb23/data/events
    21. # Describe the sink
    22. events.sinks.eventsSink.type=logger
    23. # Bind the source and sink to the channel
    24. events.sources.eventsSource.channels=eventsChannel
    25. events.sinks.eventsSink.channel=eventsChannel

    (3)sink输出指向kafka,topic

    [root@kb129 myconf2]# vim ./events-flume-kafka.conf

    1. events.sources=eventsSource
    2. events.channels=eventsChannel
    3. events.sinks=eventsSink
    4. # Describe/configure the source
    5. events.sources.eventsSource.type=spooldir
    6. events.sources.eventsSource.spoolDir=/opt/kb23/flumelogfile/events
    7. #反序列化类型为LINE
    8. events.sources.eventsSource.deserializer=LINE
    9. #反序列化 每行的最大长度
    10. events.sources.eventsSource.deserializer.maxLineLength=32000
    11. events.sources.eventsSource.includePattern=events_[0-9]{4}-[0-9]{2}-[0-9]{2}.csv
    12. #正则过滤器
    13. events.sources.eventsSource.interceptors=head_filter
    14. events.sources.eventsSource.interceptors.head_filter.type=regex_filter
    15. events.sources.eventsSource.interceptors.head_filter.regex=^event_id*
    16. events.sources.eventsSource.interceptors.head_filter.excludeEvents=true
    17. # Use a channel which buffers events in memory
    18. events.channels.eventsChannel.type=file
    19. events.channels.eventsChannel.checkpointDir=/opt/kb23/checkpoint/events
    20. events.channels.eventsChannel.dataDirs=/opt/kb23/data/events
    21. # Describe the sink
    22. events.sinks.eventsSink.type=org.apache.flume.sink.kafka.KafkaSink
    23. events.sinks.eventsSink.topic=events
    24. events.sinks.eventsSink.brokerList=192.168.142.129:9092
    25. events.sinks.eventsSink.batchSize=640
    26. # Bind the source and sink to the channel
    27. events.sources.eventsSource.channels=eventsChannel
    28. events.sinks.eventsSink.channel=eventsChannel

    2.2.7  自定义Interceptor

    )案例需求

    2.用Flume采集服务器本地日志,需要按照日志类型的不同,将不同种类的日志发往不同的分析系统。

    2)需求分析

    在实际的开发中,一台服务器产生的日志类型可能有很多种,不同类型的日志可能需要发送到不同的分析系统。此时会用到Flume拓扑结构中的Multiplexing结构,Multiplexing的原理是,根据event中Header的某个key的值,将不同的event发送到不同的Channel中,所以我们需要自定义一个Interceptor,为不同类型的event的Header中的value赋予不同的值。

    在该案例中,我们以端口数据模拟日志,以数字(单个)和字母(单个)模拟不同类型的日志,我们需要自定义interceptor区分数字和字母,将其分别发往不同的分析系统(Channel)。

    重要组件:

    1ChannelSelector

    ChannelSelector的作用就是选出Event将要被发往哪个Channel。其共有两种类型,分别是Replicating(复制)和Multiplexing(多路复用)。

    ReplicatingSelector会将同一个Event发往所有的Channel,Multiplexing会根据相应的原则,将不同的Event发往不同的Channel。

    2SinkProcessor

    SinkProcessor共有三种类型,分别是DefaultSinkProcessorLoadBalancingSinkProcessorFailoverSinkProcessor

    DefaultSinkProcessor对应的是单个的Sink,LoadBalancingSinkProcessor和FailoverSinkProcessor对应的是Sink Group,LoadBalancingSinkProcessor可以实现负载均衡的功能,FailoverSinkProcessor可以错误恢复的功能。

    3)实现步骤

    (1)创建一个maven项目,并引入以下依赖。

    1. <dependency>
    2. <groupId>org.apache.flumegroupId>
    3. <artifactId>flume-ng-coreartifactId>
    4. <version>1.9.0version>
    5. dependency>

    (2)定义类并实现Interceptor接口。编写完成后打包传至flume中lib目录下

    1. import org.apache.flume.Context;
    2. import org.apache.flume.Event;
    3. import org.apache.flume.interceptor.Interceptor;
    4. import java.util.ArrayList;
    5. import java.util.List;
    6. import java.util.Map;
    7. public class InterceptorDemo implements Interceptor {
    8. private ArrayList opList;
    9. public void initialize() {
    10. opList = new ArrayList();
    11. }
    12. public Event intercept(Event event) {
    13. Map headers = event.getHeaders();
    14. final String body = new String(event.getBody());
    15. if (body.startsWith("hello")){
    16. headers.put("type","hello");
    17. } else if (body.startsWith("hi")) {
    18. headers.put("type","hi");
    19. } else {
    20. headers.put("type","other");
    21. }
    22. return event;
    23. }
    24. public List intercept(List events) {
    25. opList.clear();
    26. for (Event event : events) {
    27. opList.add(intercept(event));
    28. }
    29. return opList;
    30. }
    31. public void close() {
    32. opList.clear();
    33. opList = null;
    34. }
    35. public static class Builder implements Interceptor.Builder{
    36. public Interceptor build() {
    37. return new InterceptorDemo();
    38. }
    39. public void configure(Context context) {
    40. }
    41. }
    42. }

    (3)编辑flume配置文件

    [root@kb129 myconf2]# vim ./netcat-myinterceptor.conf

    1. myinterceptor.sources=s1
    2. myinterceptor.channels=helloChannel hiChannel otherChannel
    3. myinterceptor.sinks=helloSink hiSink otherSink
    4. # Describe/configure the source
    5. myinterceptor.sources.s1.type=netcat
    6. myinterceptor.sources.s1.bind=localhost
    7. myinterceptor.sources.s1.port=7777
    8. myinterceptor.sources.s1.interceptors=myinterceptors
    9. myinterceptor.sources.s1.interceptors.myinterceptors.type=nj.zb.kb23.InterceptorDemo$Builder
    10. myinterceptor.sources.s1.selector.type=multiplexing
    11. myinterceptor.sources.s1.selector.mapping.hello=helloChannel
    12. myinterceptor.sources.s1.selector.mapping.hi=hiChannel
    13. myinterceptor.sources.s1.selector.mapping.other=otherChannel
    14. myinterceptor.sources.s1.selector.header=type
    15. # Use a channel which buffers events in memory
    16. myinterceptor.channels.helloChannel.type=memory
    17. myinterceptor.channels.hiChannel.type=memory
    18. myinterceptor.channels.otherChannel.type=memory
    19. # Describe the sink
    20. myinterceptor.sinks.helloSink.type=hdfs
    21. myinterceptor.sinks.helloSink.hdfs.fileType=DataStream
    22. myinterceptor.sinks.helloSink.hdfs.filePrefix=hellocontent
    23. myinterceptor.sinks.helloSink.hdfs.fileSuffix=.txt
    24. myinterceptor.sinks.helloSink.hdfs.path=hdfs://kb129:9000/kb23hello/
    25. #myinterceptor.sinks.hiSink.type=org.apache.flume.sink.kafka.KafkaSink
    26. #myinterceptor.sinks.hiSink.kafka.topic=hitopic
    27. #myinterceptor.sinks.hiSink.kafka.bootstrap.servers=192.168.142.129:9092
    28. #myinterceptor.sinks.hiSink.kafka.producer.acks=1
    29. #myinterceptor.sinks.hiSink.kafka.flumeBatchSize=640
    30. myinterceptor.sinks.hiSink.type=org.apache.flume.sink.kafka.KafkaSink
    31. myinterceptor.sinks.hiSink.topic=hitopic
    32. myinterceptor.sinks.hiSink.brokerList=192.168.142.129:9092
    33. myinterceptor.sinks.otherSink.type=logger
    34. # Bind the source and sink to the channel
    35. myinterceptor.sources.s1.channels=helloChannel hiChannel otherChannel
    36. myinterceptor.sinks.helloSink.channel=helloChannel
    37. myinterceptor.sinks.hiSink.channel=hiChannel
    38. myinterceptor.sinks.otherSink.channel=otherChannel

    3.Flume事务

  • 相关阅读:
    l8-d7 实现TCP通信
    HTML做一个简单漂亮的宠物网页(纯html代码)
    【FAQ】安防视频监控平台EasyNVR无法控制云台,该如何解决?
    [附源码]Python计算机毕业设计Django小区疫情事件处理系统
    电机控制从入门到吹牛
    Review of Algorithm (HITSZ)
    05.SpringBoot依赖管理你学会了吗
    数据库编程
    化学制品制造业数智化供应链管理系统:建立端到端供应链采购一体化平台
    Ubuntu22.04 vnc远程黑屏
  • 原文地址:https://blog.csdn.net/weixin_63713552/article/details/133243511