bin/kafka-server-start.sh config/server.properties

bin/kafka-topics.sh --create --zookeeper vm: 2181 --replication-factor 1 --partitions 1 --topic test

#define c1agent1.channels.c1.type = memoryagent1.channels.c1.capacity = 2000000agent1.channels.c1.transactionCapacity = 100#define c1 end#define c2agent1.channels.c2.type = memoryagent1.channels.c2.capacity = 2000000agent1.channels.c2.transactionCapacity = 100#define c2 end#define source monitor a fileagent1.sources.avro-s.type = execagent1.sources.avro-s.command = tail -f -n+1 /usr/local/hong/apache-tomcat-8.0.32/logs/localhost_access_log.2016-02-19.txtagent1.sources.avro-s.channels = c1 c2agent1.sources.avro-s.threads = 5# send to hadoopagent1.sinks.log-hdfs.channel = c1agent1.sinks.log-hdfs.type = hdfsagent1.sinks.log-hdfs.hdfs.path = hdfs://vm:9000/flumeagent1.sinks.log-hdfs.hdfs.writeFormat = Textagent1.sinks.log-hdfs.hdfs.fileType = DataStreamagent1.sinks.log-hdfs.hdfs.rollInterval = 0agent1.sinks.log-hdfs.hdfs.rollSize = 1000000agent1.sinks.log-hdfs.hdfs.rollCount = 0agent1.sinks.log-hdfs.hdfs.batchSize = 1000agent1.sinks.log-hdfs.hdfs.txnEventMax = 1000agent1.sinks.log-hdfs.hdfs.callTimeout = 60000agent1.sinks.log-hdfs.hdfs.appendTimeout = 60000#send to kafakaagent1.sinks.log-sink2.type = org.apache.flume.sink.kafka.KafkaSinkagent1.sinks.log-sink2.topic = testagent1.sinks.log-sink2.brokerList = vm:9092agent1.sinks.log-sink2.requiredAcks = 1agent1.sinks.log-sink2.batchSize = 20agent1.sinks.log-sink2.channel = c2# Finally, now that we‘ve defined all of our components, tell# agent1 which ones we want to activate.agent1.channels = c1 c2agent1.sources = avro-sagent1.sinks = log-hdfs log-sink2

./start-dfs.sh

./zkServer.sh start

 flume-ng agent -c conf -f dh.conf -n agent1 -Dflume.root.logger=INFO,console

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning