flume采集案例

短命女 2022-03-15 11:04 256阅读 0赞

**1、采集目录到HDFS**

采集需求：某服务器的某特定目录下，会不断产生新的文件，每当有新文件出现，就需要把文件采集到HDFS中去

根据需求，首先定义以下3大要素

1.  采集源，即source——监控文件目录 :  spooldir
2.  下沉目标，即sink——HDFS文件系统  :  hdfs sink
3.  source和sink之间的传递通道——channel，可用file channel 也可以用内存channel

配置文件编写：

\#定义三大组件的名称

agent1.sources = source1

agent1.sinks = sink1

agent1.channels = channel1

\# 配置source组件

agent1.sources.source1.type = spooldir

agent1.sources.source1.spoolDir = /home/hadoop/logs/

agent1.sources.source1.fileHeader = false

\#配置拦截器

agent1.sources.source1.interceptors = i1

agent1.sources.source1.interceptors.i1.type = host

agent1.sources.source1.interceptors.i1.hostHeader = hostname

\# 配置sink组件

agent1.sinks.sink1.type = hdfs

agent1.sinks.sink1.hdfs.path =hdfs://hdp-node-01:9000/weblog/flume-collection/%y-%m-%d/%H-%M

agent1.sinks.sink1.hdfs.filePrefix = access\_log

agent1.sinks.sink1.hdfs.maxOpenFiles = 5000

agent1.sinks.sink1.hdfs.batchSize= 100

agent1.sinks.sink1.hdfs.fileType = DataStream

agent1.sinks.sink1.hdfs.writeFormat =Text

agent1.sinks.sink1.hdfs.rollSize = 102400

agent1.sinks.sink1.hdfs.rollCount = 1000000

agent1.sinks.sink1.hdfs.rollInterval = 60

\#agent1.sinks.sink1.hdfs.round = true

\#agent1.sinks.sink1.hdfs.roundValue = 10

\#agent1.sinks.sink1.hdfs.roundUnit = minute

agent1.sinks.sink1.hdfs.useLocalTimeStamp = true

\# Use a channel which buffers events in memory

agent1.channels.channel1.type = memory

agent1.channels.channel1.keep-alive = 120

agent1.channels.channel1.capacity = 500000

agent1.channels.channel1.transactionCapacity = 600

\# Bind the source and sink to the channel

agent1.sources.source1.channels = channel1

agent1.sinks.sink1.channel = channel1

Channel参数解释：

capacity：默认该通道中最大的可以存储的event数量

trasactionCapacity：每次最大可以从source中拿到或者送到sink中的event数量

keep-alive：event添加到通道中或者移出的允许时间

**2、采集文件到HDFS**

采集需求：比如业务系统使用log4j生成的日志，日志内容不断增加，需要把追加到日志文件中的数据实时采集到hdfs

根据需求，首先定义以下3大要素

1.  采集源，即source——监控文件内容更新 :  exec  ‘tail -F file’
2.  下沉目标，即sink——HDFS文件系统  :  hdfs sink
3.  Source和sink之间的传递通道——channel，可用file channel 也可以用 内存channel

配置文件编写：

agent1.sources = source1

agent1.sinks = sink1

agent1.channels = channel1

\# Describe/configure tail -F source1

agent1.sources.source1.type = exec

agent1.sources.source1.command = tail -F /home/hadoop/logs/access\_log

agent1.sources.source1.channels = channel1

\#configure host for source

agent1.sources.source1.interceptors = i1

agent1.sources.source1.interceptors.i1.type = host

agent1.sources.source1.interceptors.i1.hostHeader = hostname