/*官方提供的kafka sink*/
a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = mytopic
a1.sinks.k1.kafka.bootstrap.servers = localhost:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
a1.sinks.ki.kafka.producer.compression.type = snappy




/**在kafka中創建一個名為 flumetest的topic*/
bin/kafka-console-consumer.sh --zookeeper ?mpc5:2181,mpc6:2181,mpc7 ?--from-beginning --topic flumetest
/**kafka中在控制臺上消費消息隊列的啟動方式*/
bin/kafka-console-consumer.sh --zookeeper ?mpc5:2181,mpc6:2181,mpc7 ?--from-beginning --topic flumetest






/*比較好用的kafka sink 但是不可靠*/
producer.sinks.r.type = org.apache.flume.plugins.KafkaSink
producer.sinks.r.metadata.broker.list=dn1:9092,dn2:9092,dn3:9092
producer.sinks.r.partition.key=0
producer.sinks.r.partitioner.class=org.apache.flume.plugins.SinglePartition
producer.sinks.r.serializer.class=kafka.serializer.StringEncoder
producer.sinks.r.request.required.acks=0
producer.sinks.r.max.message.size=1000000
producer.sinks.r.producer.type=sync
producer.sinks.r.custom.encoding=UTF-8
producer.sinks.r.custom.topic.name=test




/*一個配置文件開始*/




a1.sources = r1
a1.channels = c1
a1.sinks =k1








#source
a1.sources.r1.type = netcat
a1.sources.r1.bind = mpc1
a1.sources.r1.port = 11203
a1.sources.r1.channels = c1






# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100




#sink
a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = flumetest
a1.sinks.k1.kafka.bootstrap.servers = mpc5:9092,mpc6:9092,mpc7:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 0
a1.sinks.k1.kafka.producer.compression.type = snappy
a1.sinks.k1.kafka.producer.zk.connect=mpc5:2181,mpc6:2181,mpc7:2181
a1.sinks.k1.kafka.producer.serializer.class=kafka.serializer.StringEncoder






/*一個配置文件結束*/






\x0D




/*tcp采集源*/


a1.sources = r1
a1.channels = c1
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 5140
a1.sources.r1.host = localhost
a1.sources.r1.channels = c1
























/*多個采集源的flume，分別是tcp和udp；sink到kafka中 開始*/
a1.sources = r1 r2
a1.channels = c1
a1.sinks =k1








#source
a1.sources.r1.type = multiport_syslogtcp
a1.sources.r1.channels = c1
a1.sources.r1.host = mpc1
a1.sources.r1.ports = 10001 10002 10003
a1.sources.r1.portHeader = port


a1.sources.r2.type = syslogudp
a1.sources.r2.port = 5140
a1.sources.r2.host = mpc1
a1.sources.r2.channels = c1






# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100




#sink
a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = flumetest
a1.sinks.k1.kafka.bootstrap.servers = mpc5:9092,mpc6:9092,mpc7:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 0
a1.sinks.k1.kafka.producer.compression.type = snappy
a1.sinks.k1.kafka.producer.zk.connect=mpc5:2181,mpc6:2181,mpc7:2181
a1.sinks.k1.kafka.producer.serializer.class=kafka.serializer.StringEncoder


/*多個采集源的flume，分別是tcp和udp；sink到kafka中 結束*/






/*啟動flume的方法*/
bin/flume-ng agent -n a1 -c conf -f conf/flume-conf.properties




/*刪除一個topic在zookeeper上的元數據*/
bin/kafka-run-class.sh kafka.admin.DeleteTopicCommand --topic mytri1 --zookeeper mpc5:2181




/**flume到kafka利用id來進行分區的配置*/


a1.sources = r1 r2
a1.channels = c1
a1.sinks =k1








#source
a1.sources.r1.type = multiport_syslogtcp
a1.sources.r1.channels = c1
a1.sources.r1.host = mpc1
a1.sources.r1.ports = 10001 10002 10003
a1.sources.r1.portHeader = port
a1.sources.r1.interceptors = i2
a1.sources.r1.interceptors.i2.type=org.apache.flume.sink.solr.morphline.UUIDInterceptor$Builder
a1.sources.r1.interceptors.i2.headerName=key
a1.sources.r1.interceptors.i2.preserveExisting=false






a1.sources.r2.type = syslogudp
a1.sources.r2.port = 5140
a1.sources.r2.host = mpc1
a1.sources.r2.channels = c1
a1.sources.r2.interceptors = i2
a1.sources.r2.interceptors.i2.type=org.apache.flume.sink.solr.morphline.UUIDInterceptor$Builder
a1.sources.r2.interceptors.i2.headerName=key
a1.sources.r2.interceptors.i2.preserveExisting=false




# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100




#sink
a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = mytri3
a1.sinks.k1.kafka.bootstrap.servers = mpc5:9092,mpc6:9092,mpc7:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 0
a1.sinks.k1.kafka.producer.zk.connect=mpc5:2181,mpc6:2181,mpc7:2181
a1.sinks.k1.kafka.producer.serializer.class=kafka.serializer.StringEncoder

總結

以上是生活随笔為你收集整理的flume通过tcp/udp采集数据并存到kafka配置及操作方式的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。