flume ng1.6 +kafka sink

(安西)

　　官方出了kafka sink 。晚上找的都配置不对。还是看源码搞定
　　###source
　　a1.sources = sysSrc
　　a1.sources.sysSrc.type=syslogudp
　　a1.sources.sysSrc.bind=0.0.0.0
　　a1.sources.sysSrc.port=3333
　　a1.sources.sysSrc.channels=fileChannel
　　#a1.sources.log4jSrc.type =
　　#a1.sources.log4jSrc.channels = fileChannel
　　#### channel
　　a1.channels = fileChannel
　　a1.channels.fileChannel.type =memory
　　a1.channels.fileChannel.capacity=100
　　#### sink
　　a1.sinks = kafkaSink
　　a1.sinks.kafkaSink.channel=fileChannel
　　a1.sinks.kafkaSink.type=org.apache.flume.sink.kafka.KafkaSink
　　a1.sinks.kafkaSink.brokerList=10.73.195.136:9092
　　a1.sinks.kafkaSink.custom.partition.key=kafkaPartition
　　a1.sinks.kafkaSink.topic=cmjTopic
　　a1.sinks.kafkaSink.serializer.class=kafka.serializer.StringEncoder
　　启动
　　bin/flume-ng agent -c conf/ -f conf/a1.properties -Dflume.root.logger=DEBUG,console -n a1
　　客户端控制台
　　echo "aaaaaaaaa" |nc -u 127.0.0.1 3333
　　agent 控制台看到
　　2015-08-19 16:47:58,342 (SinkRunner-PollingRunner-DefaultSinkProcessor) [DEBUG - org.apache.flume.sink.kafka.KafkaSink.process(KafkaSink.java:114)] {Event} cmjTopic : null : aaaaaaaaa
　　打开kafkaConsumer
　　bin/kafka-console-consumer.sh --zookeeper 127.0.0.1:2181 --topic cmjTopic --from-beginning
　　看到
　　aaaaaaa输出