使用flume收集文件数据传输至kafka

曾经终败给现在 2023-02-17 11:27 6阅读 0赞

### 不做过多介绍  直接实现目的 ###

kafka和flume的包就自行百度即可,kafka的安装教程:[https://blog.csdn.net/qq\_41594146/article/details/100153434][https_blog.csdn.net_qq_41594146_article_details_100153434]

### flume的话直接下载后解压即可用   ###

### 直接上解释的配置文件: ###

#agent name :a1    #给代理取的名字
    a1.sources = r1    #sources认定下方带r1
    a1.sinks = k1      #理解同上
    a1.channels = c1   #理解同上
    
    # Flume source
    a1.sources.r1.type = exec   #监听的类型
    a1.sources.r1.command = tail -f /home/testdata.txt  #监听时使用的指令  这个可自行修改成你需要的指令 加 |grep  可实现过滤效果
    a1.sources.r1.shell=/bin/sh -c   
    a1.sources.r1.batchSize = 1000
    a1.sources.r1.batchTimeout = 3000
    a1.sources.r1.channels = c1
    
    # Flume channel 
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000000
    a1.channels.c1.transactionCapacity = 100000
    a1.channels.c1.byteCapacityBufferPercentage = 10
    #a1.channels.c1.byteCapacity = 800000
    
    # Flume sink
    a1.sinks.k1.channel = c1
    a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
    a1.sinks.k1.topic = test
    a1.sinks.k1.brokerList = 192.168.79.137:9092   #kafka的配置  如果是集群 用逗号隔开
    a1.sinks.k1.requiredAcks = 1
    a1.sinks.k1.batchSize = 2000

### 运行flume: ###

nohup bin/flume-ng agent -c conf -f conf/flume2kafka.properties -n a1 -Dflume.root.logger=INFO,console &
    
    # -f后带的是指定的配置文件名  -n 后面带的是代理的名字  上方的配置文件用的是a1所以这里用a1  再后面带的是运行时的jvm参数

### 效果: ###

**模拟文件写入情况:**

![20200613232529785.png][]

**java程序消费情况:**

![20200613232542733.png][]

###  如果能可以帮到各位 可以点个赞或者关注 ###

## ##

[https_blog.csdn.net_qq_41594146_article_details_100153434]: https://blog.csdn.net/qq_41594146/article/details/100153434
[20200613232529785.png]: https://img-blog.csdnimg.cn/20200613232529785.png
[20200613232542733.png]: https://img-blog.csdnimg.cn/20200613232542733.png