用Kafka消费Flume传来的日志

kafka-topics.sh --create  --zookeeper localhost:2181  --partitions 1 --replication-factor 1   --topic weblogs

agent.sources=spooldir-source
agent.sources.spooldir-source.type=spooldir
agent.sources.spooldir-source.spoolDir=/flume/weblogs_spooldir
agent.sources.spooldir-source.deserializer.maxLineLength=20480

agent.channels=memory-channel
agent.channels.memory-channel.type = memory
agent.channels.memory-channel.capacity = 10000
agent.channels.memory-channel.transactionCapacity = 1000

agent.sinks=kafka-sink
agent.sinks.kafka-sink.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafka-sink.topic = weblogs
agent.sinks.kafka-sink.bootstrap.servers = localhost:9092
agent.sinks.kafka-sink.flumeBatchSize = 20
agent.sinks.kafka-sink.producer.acks = 1
agent.sinks.kafka-sink.producer.linger.ms = 1
agent.sinks.kafka-sink.producer.compression.type = snappy

agent.sources.spooldir-source.channel=memory-channel
agent.sinks.kafka-sink.channel=memory-channel

flume-ng  --conf  /opt/flume/conf    --conf-file   /opt/flume/conf/kafka-spool.conf    --name  agent -Dflume.root.logger=INFO,console

 kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic weblogs