这里数据源存放在hdfs上
启动项：【hadoop(主要用hdfs)，zookeeper，kafka，flume】
start-all.sh zkServer.sh start kafka-server-start.sh /opt/soft/kafka200/config/server.properties
然后再kafka上启动监听，方便观察数据的传入:
kafka-console-consumer.sh --bootstrap-server 192.168.131.200:9092 --topic events --from-beginning
准备flume的配置信息：
events.channels = c1 events.sources = s1 events.sinks = k1 events.sources.s1.type = spooldir events.sources.s1.spoolDir = /opt/mydata/events/ -- 存放数据的文件夹 events.sources.s1.interceptors = i1 -- 拦截器，去除掉数据源中的第一行 events.sources.s1.interceptors.i1.type = regex_filter events.sources.s1.interceptors.i1.regex = event_id.* events.sources.s1.interceptors.i1.excludeEvents = true events.sources.s1.deserializer.maxLineLength=204800 -- 保证每次传输最大字节，防止某一行数据过多被拆分 events.channels.c1.capacity=1000 -- 防止数据量过大，内存溢出 events.channels.c1.type = memory -- 这里我为了快，直接在内存上 events.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink events.sinks.k1.kafka.topic = events -- kafka端消息队列名称 events.sinks.k1.kafka.bootstrap.servers = 192.168.131.200:9092 events.sinks.k1.channel = c1 events.sources.s1.channels = c1
提前记录数据量：（有多少行）
wc -l /opt/mydata/events/events.csv -- 查看有多少行 wc -L /opt/mydata/events/events.csv -- 查看最长行的长度
开启flume传输：
flume-ng agent -n events -f /opt/fconf/events.conf
此时，你在之前打开的监听窗口可以看到数据的传输效果。
查看消费了多少条数据：
kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.131.200:9092 --topic events
删除队列：
首先确保/conf下的server.properties文件已经添加delete.topic.enable=true
（删除时候先关闭之前开启的对应的监听该队列的窗口）
kafka-topics.sh --zookeeper 192.168.131.200:2181 --delete --topic mydemo
查看消息队列：
kafka-topics.sh --zookeeper 192.168.131.200:2181 --list
这里所做的准备工作就是将数据从hdfs（数据源：日志文件）通过flume传进kafka，利用kafka的在内存中操作的优势（快、高效），再使用sparkStreaming来对数据进行处理。