3.2.1 数据采集
思路：
a) 配置kafka，启动zookeeper和kafka集群；
b) 创建kafka主题；
c) 启动kafka控制台消费者（此消费者只用于测试使用）；
d) 配置flume，监控日志文件；
e) 启动flume监控任务；
f) 运行日志生产脚本；
g) 观察测试。
1）启动zookeeper，kafka集群
$/opt/module/kafka/bin/kafka-server-start.sh /opt/module/kafka/config/server.properties
2）创建kafka主题
$ /opt/module/kafka/bin/kafka-topics.sh --zookeeper hadoop102:2181 --topic calllog –-create --replication-factor 1 --partitions 3
检查一下是否创建主题成功：
$ /opt/module/kafka/bin/kafka-topics.sh --zookeeper hadoop102:2181 --list
3）启动kafka控制台消费者，等待flume信息的输入
$ /opt/module/kafka/bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 –topic
calllog --from-beginning
4）配置flume(flume2kafka.conf)
define
a1.sources = r1
a1.sinks = k1
a1.channels = c1
source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F -c +0 /opt/module/calllog/calllog.csv
a1.sources.r1.shell = /bin/bash -c
sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.bootstrap.servers = hadoop101:9092,hadoop102:9092,hadoop103:9092
a1.sinks.k1.kafka.topic = calllog
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100bind
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
5）启动flume
$ /opt/module/flume/bin/flume-ng agent --conf /opt/module/flume/conf/ --name a1 --conf-file /home/tujia/calllog/flume2kafka.conf
6）运行生产日志的任务脚本，观察kafka控制台消费者是否成功显示产生的数