flume与kafka的整合

案例1:syslog-memory-kafka

将flume采集到的数据落地到kafka上,即sink是kafka(生产者身份)

vim syslog-mem-kafka.conf
# 命名个组件
a1.sources &＃61; r1
a1.sinks &＃61; k1
a1.channels &＃61; c1#source属性
a1.sources.r1.type &＃61; syslogtcp
a1.sources.r1.host&＃61;mypc01
a1.sources.r1.port&＃61;10086# 描述channel属性
a1.channels.c1.type &＃61; memory
a1.channels.c1.capacity &＃61; 1000
a1.channels.c1.transactionCapacity &＃61; 100# 描述sink属性
a1.sinks.k1.type &＃61; org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.bootstrap.servers &＃61; mypc01:9092,mypc:9092,mypc03:9092
# 主题必须提前存在
a1.sinks.k1.kafka.topic &＃61; pet
a1.sinks.k1.kafka.producer.acks &＃61; 1
a1.sinks.k1.kafka.producer.linger.ms &＃61; 1# 关联source和sink到channel
a1.sources.r1.channels &＃61; c1
a1.sinks.k1.channel &＃61; c1

启动flume

#bin/bash
/usr/local/flume/bin/flume-ng agent -c /usr/local/flume/conf \
-f /usr/local/flume/flumeconf/syslog-mem-kafka.conf \
-n a1 -Dflume.root.logger&＃61;INFO,console -Dflume.monitoring.type&＃61;http -Dflume.monitoring.port&＃61;31002


先启动消费者准备接受消息

kafka-console-consumer.sh \
--bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
--topic pet


测试

echo "aaaaa" | nc mypc01 10086


案例2 kafka-memory-hdfs

kafka的source类型从kafka集群读取数据,就是消费者身份,将数据封装成event落地到hdfs

vim kafka-mem-kafka.conf
# 命名个组件
a1.sources &＃61; r1
a1.sinks &＃61; k1
a1.channels &＃61; c1#source属性
a1.sources.r1.type &＃61; org.apache.flume.source.kafka.KafkaSource
a1.sources.r1.kafka.bootstrap.servers &＃61; mypc01:9092,mypc02:9092,mypc03:9092
a1.sources.r1.kafka.consumer.group.id&＃61;g1
a1.sources.r1.kafka.topics&＃61;pet# 描述channel属性
a1.channels.c1.type &＃61; memory
a1.channels.c1.capacity &＃61; 1000
a1.channels.c1.transactionCapacity &＃61; 100# 描述sink属性
a1.sinks.k1.type&＃61;hdfs
a1.sinks.k1.hdfs.path&＃61;hdfs://mypc01:8020/kafka/pet/%Y%m%d
a1.sinks.k1.hdfs.filePrefix&＃61;FlumeData
a1.sinks.k1.hdfs.fileSuffix &＃61; .kafka
a1.sinks.k1.hdfs.rollSize&＃61;102400
a1.sinks.k1.hdfs.rollCount &＃61; 0
#单位为s
b1001.sinks.k1.hdfs.rollInterval&＃61;60
b1001.sinks.k1.hdfs.useLocalTimeStamp &＃61; true# 关联source和sink到channel
a1.sources.r1.channels &＃61; c1
a1.sinks.k1.channel &＃61; c1

启动flume

#bin/bash
/usr/local/flume/bin/flume-ng agent -c /usr/local/flume/conf \
-f /usr/local/flume/flumeconf/kafka-mem-hdfs.conf \
-n a1 -Dflume.root.logger&＃61;INFO,console -Dflume.monitoring.type&＃61;http -Dflume.monitoring.port&＃61;31002


启动生产者,使用生产者发送消息

kafka-console-producer.sh \
--broker-list mypc01:9092,mypc02:9092,mypc03:9092 \
--topic pet


之后就可以在hdfs上看到生成的文件了.

总结

• kafka可以作为source,也可以作为sink