背景：需要實時讀取log日志文件中的記錄到kafka
1.zookeeper服務需要開啟，查看zookeeper的狀態，（zookeeper的安裝及啟動過程可查看 https://www.cnblogs.com/cstark/p/14573395.html）
[root@master kafka_2.11-0.11]# /opt/soft/zookeeper-3.4.13/bin/zkServer.sh status ZooKeeper JMX enabled by default Using config: /opt/soft/zookeeper-3.4.13/bin/../conf/zoo.cfg Mode: follower
2.kafka服務需要開啟
/opt/soft/kafka_2.11-0.11/bin/kafka-server-start.sh /opt/soft/kafka_2.11-0.11/config/server.properties
3.啟動flume 的配置文件
啟動flume：bin/flume-ng agent --conf conf --conf-file ./conf/job/file_to_hdfs.conf --name a1
a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = exec a1.sources.r1.command = tail -F /opt/data/mall/16/mall.log # 需要監控的文件 # Describe the sink #a1.sinks.k1.type = logger a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.topic = test a1.sinks.k1.brokerList = master:9092 a1.sinks.k1.requiredAcks = 1 a1.sinks.k1.batchSize = 20 # Use a channel which buffers events in memory a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # Bind the source and sink to the channel a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1
4.查看kafka對應的topic里面數據是否有同步過來
/opt/soft/kafka_2.11-0.11/bin/kafka-console-consumer.sh --bootstrap-server master:9092 --from-beginning --topic test
