Flume的入门及小案例

我不是女神ヾ 2023-10-10 08:54 42阅读 0赞

# Flume #

*  **概念**:

Flume是一种分布式、可靠且可用的服务，用于高效地收集、聚合和移动大量日志数据。它具有基于流数
    据流的简单灵活的体系结构。它具有健壮性和容错性，具有可调的可靠性机制和许多故障转移和恢复机制
    。它使用一个简单的可扩展数据模型，允许在线分析应用程序。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70]

*  **组件：**

source：数据源组件,用于读取相应数据,并将数据传到channel中
    channel:管道，用于连接source和sink
    sink:数据下沉组件,用于将channel中的数据持久化到对应的文件系统中或者流中
    agent：flume的运行单元.里面必须包含一个或者多个source,channel,sink,运行在单个jvm中
    event：事件，是数据的描述
    interceptor:拦截器,作用于source阶段,用于过滤数据
    selectorer:选择器,作用于source阶段,默认是replicating，也就是复用功能,mutiplxing
    groupsinks：sink组,用于将多个sink选择sinkf
    
    常用的组件:
    source常用的组件:exec、avro、source、spooling、dirctory、kafka、netcat、http、自定义
    channel常用的组件:memory、file、kafka、jdbc
    sink常用组件:avro、hdfs、hive、hbase、kafka

## 6.案例 ##

6.1 exec -->memory -->logger

在/home下创建一个f_log随便写入
    在conf下创建一个test文件夹,创建first.job进行配置
    
    #定义source|channel|sink组件
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    #配置r1的属性
    a1.sources.r1.type = exec
    a1.sources.r1.command = tail -F /home/f_log
    
    #配置sinks的属性
    a1.sinks.k1.type = logger
    
    #配置channel的属性
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 10000
    a1.channels.c1.transactionCapacity = 10000
    
    #绑定source与sink于channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1
    
    启动agent：
    flume-ng agent -c conf/  -f ./conf/test/first.conf -n a1 -Dflume.root.logger=INFO,console
    
    测试:
    echo "123" >> /home/f_log

6.2 exec -->memory -->hdfs

在conf下创建一个test文件夹,创建exememhdf.conf进行配置
    
    #定义source|channel|sink组件
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    #配置r1的属性
    a1.sources.r1.type = exec
    a1.sources.r1.command = tail -F /home/f_log
    
    #配置sinks的属性
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = /flume/events/dt=%y-%m-%d
    a1.sinks.k1.hdfs.filePrefix = events-
    a1.sinks.k1.hdfs.round = true
    a1.sinks.k1.hdfs.roundValue = 10
    a1.sinks.k1.hdfs.writeFormat=Text
    a1.sinks.k1.hdfs.roundUnit=DataStream
    a1.sinks.k1.hdfs.roundUnit = minute
    a1.sinks.k1.hdfs.useLocalTimeStamp=true
    
    
    #配置channel的属性
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 10000
    a1.channels.c1.transactionCapacity = 10000
    
    #绑定source与sink于channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1
    
    启动agent:
    flume-ng agent -c conf/ -f conf/test/exememhdf.conf -n a1 -Dflume.root.logger=INFO,console
    
    测试:
    echo "123" >> /home/f_log

spooldir -->memory -->logger

创建/home/flumedata/spool1 目录
    在conf下创建一个test文件夹,创建spomemlog.conf进行配置
    
    a1.sources=r1
    a1.channels=c1
    a1.sinks=s1
    
    a1.sources.r1.type=spoolDir
    a1.sources.r1.spoolDir=/home/flumedata/spool1
    a1.sources.r1.fileHeader=true
    a1.sources.r1.fileHeaderKey=file
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    a1.channels.c1.keep-alive=3
    a1.channels.c1.byteCapacityBufferPercentage = 20
    a1.channels.c1.byteCapacity = 800000
    
    
    a1.sinks.s1.type = logger
    
    a1.sources.r1.channels=c1
    a1.sinks.s1.channel=c1
    
    启动agent:
    flume-ng agent -c ./conf -f ./conf/test/spomemlog.conf -n a1 -Dflume.root.lgger=INFO,console
    
    测试:
    echo "123" >> /home/flumedata/spool1/a.txt
    echo "456" >> /home/flumedata/spool1/b.txt
    因为spool读取的是文件,读完之后默认将原文件重命名为xxx.COMPLETED,所以每次添加文件的名称不能相同,否则会卡住

syslogtcp -->memory -->logger

在conf下创建一个test文件夹,创建sysmemlog.conf进行配置
    
    a1.sources=r1
    a1.channels=c1
    a1.sinks=s1
    
    a1.sources.r1.type=syslogtcp
    a1.sources.r1.port=6666
    a1.sources.r1.host=hadoop111
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    a1.channels.c1.keep-alive=3
    a1.channels.c1.byteCapacityBufferPercentage = 20
    a1.channels.c1.byteCapacity = 800000
    
    a1.sinks.s1.type = logger
    
    a1.sources.r1.channels=c1
    a1.sinks.s1.channel=c1
    
    启动agent:
    flume-ng agent -c ./conf -f ./conf/test/sysmemlog.conf -n a1 -Dflume.root.lgger=INFO,console
    
    测试：
    需要先安装nc命令   
    yum install nc
    
    echo "hello world" | nc hadoop111 6666

http -->memory -->logger

在conf下创建一个test文件夹,创建httmemlog.conf进行配置
    
    a1.sources=r1
    a1.channels=c1
    a1.sinks=s1
    
    a1.sources.r1.type=org.apache.flume.source.http.HTTPSource
    a1.sources.r1.port=6666
    a1.sources.r1.bind=hadoop111
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    a1.channels.c1.keep-alive=3
    a1.channels.c1.byteCapacityBufferPercentage = 20
    a1.channels.c1.byteCapacity = 800000
    
    a1.sinks.s1.type = logger
    
    a1.sources.r1.channels=c1
    a1.sinks.s1.channel=c1
    
    启动agent:
    flume-ng agent -c ./conf -f ./conf/test/httmemlog.conf -n a1 -Dflume.root.lgger=INFO,console
    
    测试:
    curl -X POST -d '[{"headers":{"time":"2008-08-08"},"body":"this is http test"}]' http://hadoop111:6666
    
    //Http Source默认的Hander类型是JSONHandler 所以测试需要传入json格式的数据

exec -->file -->hdfs

在conf下创建一个test文件夹,创建exefilhdf.conf进行配置
    
    a1.sources=r1
    a1.channels=c1
    a1.sinks=s1
    
    a1.sources.r1.type=exec
    a1.sources.r1.command= tail -f /home/flumedata/exedata
    
    a1.channels.c1.type=file
    a1.channels.c1.checkpointDir=/home/flumedata/checkpoint
    a1.channels.c1.dataDirs=/home/flumedata/data
    
    a1.sinks.s1.type = hdfs
    a1.sinks.s1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S
    a1.sinks.s1.hdfs.filePrefix = zxh-
    a1.sinks.s1.hdfs.fileSuffix=.log
    a1.sinks.s1.hdfs.inUseSuffix=.tmp
    a1.sinks.s1.hdfs.rollInterval=2
    a1.sinks.s1.hdfs.rollSize=1024
    a1.sinks.s1.hdfs.fileType=DataStream
    a1.sinks.s1.hdfs.writeFormat=Text
    a1.sinks.s1.hdfs.round = true
    a1.sinks.s1.hdfs.roundValue = 1
    a1.sinks.s1.hdfs.roundUnit = second
    a1.sinks.s1.hdfs.useLocalTimeStamp=true
    
    
    a1.sources.r1.channels=c1
    a1.sinks.s1.channel=c1
    
    启动agent:
    flume-ng agent -c ./conf -f ./conf/test/exefilhdf.conf -n a1 -Dflume.root.lgger=INFO,console
    
    测试:
    echo "123" >> /home/flumedata/exedata

exec + interceptors -->memory -->hdfs  
时间戳拦截器  
静态拦截器  
主机拦截器

在conf下创建一个test文件夹,创建exeintmemhdf.conf进行配置
    
    a1.sources=r1
    a1.channels=c1
    a1.sinks=s1
    
    a1.sources.r1.type=exec
    a1.sources.r1.command= tail -f /home/flumedata/exedata
    
    a1.sources.r1.interceptors = i1 i2 i3
    a1.sources.r1.interceptors.i1.type = timestamp
    a1.sources.r1.interceptors.i1.preserveExisting=true
    
    a1.sources.r1.interceptors.i2.type = host
    a1.sources.r1.interceptors.i2.hostHeader = hostname
    a1.sources.r1.interceptors.i2.preserveExisting=true
    
    a1.sources.r1.interceptors.i3.type = static
    a1.sources.r1.interceptors.i3.key = city
    a1.sources.r1.interceptors.i3.value = NEW_YORK
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    a1.channels.c1.keep-alive=3
    a1.channels.c1.byteCapacityBufferPercentage = 20
    a1.channels.c1.byteCapacity = 800000
    
    a1.sinks.s1.type = hdfs
    a1.sinks.s1.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S
    a1.sinks.s1.hdfs.filePrefix = %{hostname}-
    a1.sinks.s1.hdfs.fileSuffix=.log
    a1.sinks.s1.hdfs.inUseSuffix=.tmp
    a1.sinks.s1.hdfs.rollInterval=2
    a1.sinks.s1.hdfs.rollSize=1024
    a1.sinks.s1.hdfs.fileType=DataStream
    a1.sinks.s1.hdfs.writeFormat=Text
    a1.sinks.s1.hdfs.round = true
    a1.sinks.s1.hdfs.roundValue = 1
    a1.sinks.s1.hdfs.roundUnit = second
    a1.sinks.s1.hdfs.useLocalTimeStamp=true
    
    a1.sources.r1.channels=c1
    a1.sinks.s1.channel=c1
    
    启动agent:
    flume-ng agent -c ./conf -f ./conf/test/exeintmemhdf.conf -n a1 -Dflume.root.lgger=INFO,console
    
    测试:
    echo "123" >> /home/flumedata/exedata

exec + interceptors -->memory -->logger  
正则拦截器

在conf下创建一个test文件夹,创建exeint2memlog.conf进行配置
    
    a1.sources=r1
    a1.channels=c1
    a1.sinks=s1
    
    a1.sources.r1.type=exec
    a1.sources.r1.command= tail -f /home/flumedata/exedata
    a1.sources.r1.interceptors = i1
    a1.sources.r1.interceptors.i1.type = regex_filter
    #定义正则,以数字开头
    a1.sources.r1.interceptors.i1.regex=^[0-9].*$
    a1.sources.r1.interceptors.i1.excludeEvents=false
    #false匹配正则
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    a1.channels.c1.keep-alive=3
    a1.channels.c1.byteCapacityBufferPercentage = 20
    a1.channels.c1.byteCapacity = 800000
    
    a1.sinks.s1.type = logger
    
    a1.sources.r1.channels=c1
    a1.sinks.s1.channel=c1
    
    启动agent:
    flume-ng agent -c ./conf -f ./conf/test/exeint2memlog.conf -n a1 -Dflume.root.lgger=INFO,console
    
    测试:
    - 数字开头:
    echo "123" >> /home/flumedata/exedata
    日志打印 可以被收集
    - 非数字开头:
    echo "abc123" >> /home/flumedata/exedata
    日志不打印 不收集

exec +selector–>memory -->hdfs  
复制选择器

在conf下创建一个test文件夹,创建exeselmemhdf.conf进行配置
    
    a1.sources=r1
    a1.channels=c1 c2
    a1.sinks=s1 s2
    
    a1.sources.r1.type=exec
    a1.sources.r1.command= tail -f /home/flumedata/exedata
    a1.sources.r1.selector.type = replicating
    a1.sources.r1.selector.optional = c2
    
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    a1.channels.c1.keep-alive=3
    a1.channels.c1.byteCapacityBufferPercentage = 20
    a1.channels.c1.byteCapacity = 800000
    
    a1.channels.c2.type=memory
    a1.channels.c2.capacity=1000
    a1.channels.c2.transactionCapacity=100
    a1.channels.c2.keep-alive=3
    a1.channels.c2.byteCapacityBufferPercentage = 20
    a1.channels.c2.byteCapacity = 800000
    
    a1.sinks.s1.type = logger
    
    a1.sinks.s2.type = hdfs
    a1.sinks.s2.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S
    a1.sinks.s2.hdfs.filePrefix = event-
    a1.sinks.s2.hdfs.fileSuffix=.log
    a1.sinks.s2.hdfs.inUseSuffix=.tmp
    a1.sinks.s2.hdfs.rollInterval=2
    a1.sinks.s2.hdfs.rollSize=1024
    a1.sinks.s2.hdfs.fileType=DataStream
    a1.sinks.s2.hdfs.writeFormat=Text
    a1.sinks.s2.hdfs.round = true
    a1.sinks.s2.hdfs.roundValue = 1
    a1.sinks.s2.hdfs.roundUnit = second
    a1.sinks.s2.hdfs.useLocalTimeStamp=true
    
    a1.sources.r1.channels=c1 c2
    a1.sinks.s1.channel=c1
    a1.sinks.s2.channel=c2
    
    启动agent:
    flume-ng agent -c ./conf -f ./conf/test/exeselmemhdf.conf -n a1 -Dflume.root.lgger=INFO,console
    
    测试:
    echo "123456" >> /home/flumedata/exedata

exec+selector -->memory -->hdfs  
复用选择器

在conf下创建一个test文件夹,创建httselmemhdf.conf进行配置
    
    a1.sources=r1
    a1.channels=c1 c2
    a1.sinks=s1 s2
    
    a1.sources.r1.type=http
    a1.sources.r1.port=6666
    a1.sources.r1.bind=hadoop111
    a1.sources.r1.selector.type = multiplexing
    a1.sources.r1.selector.header = status
    a1.sources.r1.selector.mapping.CZ = c1
    a1.sources.r1.selector.mapping.US = c2
    a1.sources.r1.selector.default = c1
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    a1.channels.c1.keep-alive=3
    a1.channels.c1.byteCapacityBufferPercentage = 20
    a1.channels.c1.byteCapacity = 800000
    
    a1.channels.c2.type=memory
    a1.channels.c2.capacity=1000
    a1.channels.c2.transactionCapacity=100
    a1.channels.c2.keep-alive=3
    a1.channels.c2.byteCapacityBufferPercentage = 20
    a1.channels.c2.byteCapacity = 800000
    
    a1.sinks.s1.type = logger
    
    a1.sinks.s2.type = hdfs
    a1.sinks.s2.hdfs.path = /flume/events/%y-%m-%d/%H%M/%S
    a1.sinks.s2.hdfs.filePrefix = zxh
    a1.sinks.s2.hdfs.fileSuffix=.log
    a1.sinks.s2.hdfs.inUseSuffix=.tmp
    a1.sinks.s2.hdfs.rollInterval=2
    a1.sinks.s2.hdfs.rollSize=1024
    a1.sinks.s2.hdfs.fileType=DataStream
    a1.sinks.s2.hdfs.writeFormat=Text
    a1.sinks.s2.hdfs.round = true
    a1.sinks.s2.hdfs.roundValue = 1
    a1.sinks.s2.hdfs.roundUnit = second
    a1.sinks.s2.hdfs.useLocalTimeStamp=true
    
    a1.sources.r1.channels=c1 c2
    a1.sinks.s1.channel=c1
    a1.sinks.s2.channel=c2
    
    启动agent:
    flume-ng agent -c ./conf -f ./conf/test/httselmemhdf.conf -n a1 -Dflume.root.lgger=INFO,console
    
    测试:
    curl -X POST -d '[{"headers":{"status":"2017-06-13"},"body":"this is default"}]' http://hadoop111:6666
    curl -X POST -d '[{"headers":{"status":"CZ"},"body":"this is CZ"}]' http://hadoop111:6666
    curl -X POST -d '[{"headers":{"status":"US"},"body":"this is US"}]' http://hadoop111:6666
    curl -X POST -d '[{"headers":{"status":"ss"},"body":"this is ss"}]' http://hadoop111:6666

**flume集群搭建**  
原理  
通过用avro sink配置一些一级代理,可以在Flume中实现这一点,它们都指向一个代理的avro源(同样,您可以在这样的场景中使用节约来源/ sink /客户机)。第二层代理的这个源将被接收的事件合并到一个单独的通道中,它被一个接收器消耗到它的最终目的地。  
1  
此处我用伪分布式举例  
配置第一个agent  
在conf下创建一个test文件夹,创建agent1.conf进行配置

a1.sources=r1
    a1.channels=c1
    a1.sinks=s1
    
    a1.sources.r1.type=netcat
    a1.sources.r1.port=6666
    a1.sources.r1.bind=hadoop111
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    a1.channels.c1.keep-alive=3
    a1.channels.c1.byteCapacityBufferPercentage = 20
    a1.channels.c1.byteCapacity = 800000
    
    a1.sinks.s1.type =avro
    a1.sinks.s1.hostname=hadoop111
    a1.sinks.s1.port=8888
    
    a1.sources.r1.channels=c1
    a1.sinks.s1.channel=c1

配置第二个agent  
在conf下创建一个test文件夹,创建agent2.conf进行配置

a1.sources=r1
    a1.channels=c1
    a1.sinks=s1
    
    a1.sources.r1.type=netcat
    a1.sources.r1.port=6667
    a1.sources.r1.bind=hadoop111
    
    a1.channels.c1.type=memory
    a1.channels.c1.capacity=1000
    a1.channels.c1.transactionCapacity=100
    a1.channels.c1.keep-alive=3
    a1.channels.c1.byteCapacityBufferPercentage = 20
    a1.channels.c1.byteCapacity = 800000
    
    a1.sinks.s1.type =avro
    a1.sinks.s1.hostname=hadoop111
    a1.sinks.s1.port=8888
    
    a1.sources.r1.channels=c1
    a1.sinks.s1.channel=c1

配置第三个agent  
在conf下创建一个test文件夹,创建agent3.conf进行配置

agent.sources=r1
    agent.channels=c1
    agent.sinks=s1
    
    agent.sources.r1.type=avro
    agent.sources.r1.port=8888
    agent.sources.r1.bind=hadoop111
    
    agent.channels.c1.type=memory
    agent.channels.c1.capacity=1000
    agent.channels.c1.transactionCapacity=100
    agent.channels.c1.keep-alive=3
    agent.channels.c1.byteCapacityBufferPercentage = 20
    agent.channels.c1.byteCapacity = 800000
    
    agent.sinks.s1.type =logger
    
    agent.sources.r1.channels=c1
    agent.sinks.s1.channel=c1
    
    agent1,agent2接收的数据都会发送给8888端口,由8888端口去处理
    启动agent:
    1.先启动avro resource所在的端口
    flume-ng agent -c ./conf -f ./conf/test/agent3.conf -n a1 -Dflume.root.lgger=INFO,console 
    2.再启动两个avro sink所在的端口
    flume-ng agent -c ./conf -f ./conf/test/agent1.conf -n a1 -Dflume.root.lgger=INFO,console 
    flume-ng agent -c ./conf -f ./conf/test/agent2.conf -n a1 -Dflume.root.lgger=INFO,console 
    
    测试1:
    echo "hello world"  | nc hadoop111 6666
    测试2:
    echo "hello world2" | nc hadoop111 6667

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20190924133742771.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NTI1OTA2,size_16,color_FFFFFF,t_70