[TOC] # 分析 采集需求：某服务器的某特定目录下，会不断产生新的文件，每当有新文件出现，就需要把文件采集到HDFS中去 根据需求，首先定义以下3大要素 * 采集源，即source——监控文件目录 : spooldir * 下沉目标，即sink——HDFS文件系统 : hdfs sink * source和sink之间的传递通道——channel，可用file channel 也可以用内存memory channel # 配置文件 ~~~ #定义三大组件的名称 agent1.sources = source1 agent1.sinks = sink1 agent1.channels = channel1 # 配置source组件 agent1.sources.source1.type = spooldir # 对那个目录进行监听上传 agent1.sources.source1.spoolDir = /root/data/ # 上传成功的文件结尾加上扩展名 agent1.sources.source1.fileSuffix=.COMPLETED # 是否使用当前文件头,无论如何加存储的绝对路径的文件名 agent1.sources.source1.fileHeader = true # 忽略所有以.tmp结尾的文件,不上传 agent1.sources.source1.ignorePattern=([^ ]*\.tmp) #配置拦截器 agent1.sources.source1.interceptors = i1 agent1.sources.source1.interceptors.i1.type = timestamp # 配置sink组件 # 类型是hdfs agent1.sinks.sink1.type = hdfs # 在hdfs上产生的目录 agent1.sinks.sink1.hdfs.path =/weblog/flume-collection/%y-%m-%d/%H-%M # 文件的前缀,在hdfs上的前缀 agent1.sinks.sink1.hdfs.filePrefix = access_log # 最大文件打开数量 agent1.sinks.sink1.hdfs.maxOpenFiles = 5000 # 批次大小,就是文件达到多少条才提交到hdfs agent1.sinks.sink1.hdfs.batchSize = 100 # 当前文件存储数据类型,还可以用压缩格式 # 文件格式：当前是SequenceFile，DataStream或CompressedStream（1）DataStream不会压缩输出文件，请不要设置codeC（2）CompressedStream需要使用可用的codeC设置hdfs.codeC agent1.sinks.sink1.hdfs.fileType = DataStream # 文件的格式类型 agent1.sinks.sink1.hdfs.writeFormat =Text # 最小冗余数,不设置为0的话,滚动策略就失效 agent1.sinks.sink1.hdfs.minBlockReplicas=0 # 达到下面的三个任何一个就按照那个标准生成一个新文件 #设置每个文件的滚动大小 agent1.sinks.sink1.hdfs.rollSize = 102400 #滚动生成的文件按行数生成,0表示和event无关 agent1.sinks.sink1.hdfs.rollCount = 100 #多久生成一个新文件,秒 agent1.sinks.sink1.hdfs.rollInterval = 10 # 整体就是每10分钟滚动生成一个目录 #开启滚动生成目录 agent1.sinks.sink1.hdfs.round = true #多长时间会创建一个新的文件夹 agent1.sinks.sink1.hdfs.roundValue = 10 #单位为分钟 agent1.sinks.sink1.hdfs.roundUnit = minute #是否使用本地时间戳 agent1.sinks.sink1.hdfs.useLocalTimeStamp = true # 管道的类型 agent1.channels.channel1.type = memory # 管道的容量,字节 agent1.channels.channel1.capacity = 500000 # 事务的类型,多少条之后source推送到channel或者channel推送到sinks agent1.channels.channel1.transactionCapacity = 600 # 多久之后将数据从source移动到channel,channel移动到sink agent1.channels.channel1.keep-alive = 120 # Bind the source and sink to the channel # 对应的source,channel进行组装 agent1.sources.source1.channels = channel1 agent1.sinks.sink1.channel = channel1 ~~~ # 测试 启动 ~~~ flume-ng agent -c conf -f fhd.conf -n agent1 -Dflume.root.logger=INFO,console ~~~ fhd.conf换成你自己写的,不同的目录加上目录 -n代表上面定义的agent的名字 启动后可以看到打印的日志,只要`/root/data/`下面有文件就会移动到hdfs 采集完的文件后面会有这个后缀`.COMPLETED` **flume的source采用spooldir时！ 目录下面不允许存放同名的文件，否则报错！** # 注意 在使用Spooling Directory Source时 1. 不要在监控目录中创建并持续修改文件 2. 上传完成的文件会以.COMPLERED结尾(fileSuffix) 3. 被监控文件夹每500毫秒扫描一次文件变动(pollDelay属性默认值) # 其他组件：Interceptor(拦截器) **用于Source的一组Interceptor**，按照预设的顺序在必要地方装饰和过滤events。 内建的Interceptors允许增加event的headers比如：时间戳、主机名、静态标记等等 定制的interceptors可以通过内省event payload（读取原始日志），实现自己的业务逻辑（很强大）