Flume -Taildir Source

﹏ヽ暗。殇╰゛Y 2022-03-09 14:24 197阅读 0赞

> Taildir Source可实时监控一批文件，并记录每个文件最新消费位置，agent进程重启后不会重复消费的问题 , 使用时建议1.8版本flume(1.8b版本解决了Taildir source的bug)  
> 功能：支持目录变化的文件，如果遇到中断，并以json数据记录目录下的每个文件的收集状态

> 重点：  
> a1.sources.r1.type = TAILDIR  
> a1.sources.r1.channels = c1  
> a1.sources.r1.positionFile = /home/web\_admin/opt/v2\_flume-apache170/logfile\_stats/x1/taildir\_position.json 记录收集状态  
> a1.sources.r1.filegroups = f1  
> a1.sources.r1.filegroups.f1 = /home/zl/xsvr/server/xgame\_1/logs/act/zl\_war.log 目录地址，支持单个文件或正则匹配如 dir/\* 或 dir/.*log.*  
> a1.sources.r1.headers.f1.headerKey1 = value1  
> a1.sources.r1.fileHeader = true

# source的名字
    agent.sources = s1
    # channels的名字
    agent.channels = c1
    # sink的名字
    agent.sinks = r1
    
    # 指定source使用的channel
    agent.sources.s1.channels = c1
    # 指定sink使用的channel
    agent.sinks.r1.channel = c1
    
    ######## source相关配置 ########
    # source类型
    agent.sources.s1.type = TAILDIR
    # 元数据位置
    agent.sources.s1.positionFile = /Users/wangpei/tempData/flume/taildir_position.json
    # 监控的目录
    agent.sources.s1.filegroups = f1
    agent.sources.s1.filegroups.f1=/Users/wangpei/tempData/flume/data/.*log
    agent.sources.s1.fileHeader = true
    
    ######## channel相关配置 ########
    # channel类型
    agent.channels.c1.type = file
    # 数据存放路径
    agent.channels.c1.dataDirs = /Users/wangpei/tempData/flume/filechannle/dataDirs
    # 检查点路径
    agent.channels.c1.checkpointDir = /Users/wangpei/tempData/flume/filechannle/checkpointDir
    # channel中最多缓存多少
    agent.channels.c1.capacity = 1000
    # channel一次最多吐给sink多少
    agent.channels.c1.transactionCapacity = 100
    
    ######## sink相关配置 ########
    # sink类型
    agent.sinks.r1.type = org.apache.flume.sink.kafka.KafkaSink
    # brokers地址
    agent.sinks.r1.kafka.bootstrap.servers = localhost:9092
    # topic
    agent.sinks.r1.kafka.topic = testTopic3
    # 压缩
    agent.sinks.r1.kafka.producer.compression.type = snappy