flume通过tcp/udp采集数据并存到kafka配置及操作方式

2019-04-18 13:55 

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/u012613903/article/details/53022989

/*官方提供的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

发表评论

您必须 登录 才能发表留言!