kafka与flume集成小案例

谁践踏了优雅 2022-03-30 04:56 172阅读 0赞

##### Flume定位 #####

> Flume是Cloudera提供的一个高可用的，高可靠的，分布式的海量日志采集，聚合和传输的流式架构

1.  适合多个生产者
2.  适合下游消费者不多的情况(如果存在多个sink,就需要多个channel，多个memchannel会占用大量的内存)
3.  适合数据安全性不高的操作(flume没有备份机制，数据安全性较低)
4.  flume可以简单的和hadoop生态圈进行对接（可以和HDFS进行对接，写conf文件即可）

##### kafka定位 #####

> Kafka是一个分布式消息队列，发送消息者成为producer，消息接收者成为Consumer，kafka集群由多个kafka实例组成，成为broker

1.  适合下游消费者较多的情况（消费者只需要订阅即可）
2.  适合数据安全性要求较高的操作（支持replication）

> 由于flume和kafka两个框架的特点，所以常常会把两者进行结合使用。 flume->kafka

#flume与kafka集成
    #定义组件名称
    a1.sources=r1
    a1.channels=c1
    a1.sinks=s1
    
    #定义source的类型和属性
    #定义类型为exec类型，监控某个文件的变化
    a1.sources.r1.type=exec
    #使用tail -F 实时监控文件的变化 -c +0 指定从每个文件的第k个字节进行输出
    a1.sources.r1.command=tail -F -c +0 /opt/module/datas/flume.log
    #指定shell命令解释器，其实可以不指定，会使用默认的，也是一样的效果
    a1.sources.r1.shell = /bin/bash -c
    
    #定义sink的类型和属性
    #类型指定为kafkaSink==>官网中可找到
    a1.sinks.s1.type=org.apache.flume.sink.kafka.KafkaSink
    #指定kafka集群的broker的地址
    a1.sinks.s1.kafka.bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop105:9092
    #指定主题
    a1.sinks.s1.kafka.topic=first
    #指定批处理大小
    a1.sinks.k1.kafka.flumeBatchSize = 20
    #指定确认ack信号
    a1.sinks.k1.kafka.producer.acks = 1
    #指定延时时间
    a1.sinks.k1.kafka.producer.linger.ms = 1
    
    #指定channel类型
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    # bind --进行绑定和拼接
    a1.sources.r1.channels = c1
    a1.sinks.s1.channel = c1

###### 执行flume程序，然后向文件中追加数据，在idea中查看运行的消费者程序执行结果 ######

bin/flume-ng agent -c conf/ -n a1 -f jobs/flume-kafka.conf