Flume和Kafka完成实时数据的采集
写在前面
Flume和Kafka在生产环境中，一般都是结合起来使用的。可以使用它们两者结合起来收集实时产生日志信息，这一点是很重要的。如果，你不了解flume和kafka，你可以先查看我写的关于那两部分的知识。再来学习，这部分的操作，也是可以的。
实时数据的采集，就面临一个问题。我们的实时数据源，怎么产生呢？因为我们可能想直接获取实时的数据流不是那么的方便。我前面写过一篇文章，关于实时数据流的python产生器，文章地址：http://blog.csdn.net/liuge36/article/details/78596876
你可以先看一下，如何生成一个实时的数据...
思路？？如何开始呢？？
分析：我们可以从数据的流向着手，数据一开始是在webserver的，我们的访问日志是被nginx服务器实时收集到了指定的文件，我们就是从这个文件中把日志数据收集起来，即：webserver=>flume=>kafka
webserver日志存放文件位置
这个文件的位置，一般是我们自己设置的
我们的web日志存放的目录是在：
/home/hadoop/data/project/logs/access.log下面
[[email protected] logs]$ pwd
/home/hadoop/data/project/logs
[[email protected] logs]$ ls
access.log
[[email protected] logs]$ Flume
做flume,其实就是写conf文件，就面临选型的问题
source选型？channel选型？sink选型？
这里我们选择 exec source memory channel kafka sink
怎么写呢？
按照之前说的那样1234步骤
从官网中，我们可以找到我们的选型应该如何书写：
1） 配置Source
exec source
# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/hadoop/data/project/logs/access.log
a1.sources.r1.shell = /bin/sh -c2） 配置Channel
memory channel
a1.channels.c1.type = memory3） 配置Sink
kafka sink
flume1.6版本可以参照http://archive.cloudera.com/cdh5/cdh/5/flume-ng-1.6.0-cdh5.7.0/FlumeUserGuide.html#kafka-sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.brokerList = hadoop000:9092
a1.sinks.k1.topic = flume_kafka
a1.sinks.k1.batchSize = 5
a1.sinks.k1.requiredAcks =14) 把以上三个组件串起来
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1我们new一个文件叫做test3.conf
把我们自己分析的代码贴进去：
[[email protected] conf]$ vim test3.conf
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/hadoop/data/project/logs/access.log
a1.sources.r1.shell = /bin/sh -c
a1.channels.c1.type = memory
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.brokerList = hadoop000:9092
a1.sinks.k1.topic = flume_kafka
a1.sinks.k1.batchSize = 5
a1.sinks.k1.requiredAcks =1
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
这里我们先不启动，因为其中涉及到kafka的东西，必须先把kafka部署起来，，
Kafka的部署
kafka如何部署呢？？
参照官网的说法，我们首先启动一个zookeeper进程，接着，才能够启动kafka的server
Step 1: Start the zookeeper
[[email protected] ~]$
[[email protected] ~]$ jps
29147 Jps
[[email protected] ~]$ zkServer.sh start
JMX enabled by default
Using config: /home/hadoop/app/zk/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
[[email protected] ~]$ jps
29172 QuorumPeerMain
29189 Jps
[[email protected] ~]$
Step 2: Start the server
[[email protected] ~]$ kafka-server-start.sh $KAFKA_HOME/config/server.properties
#外开一个窗口，查看jps
[[email protected] ~]$ jps
29330 Jps
29172 QuorumPeerMain
29229 Kafka
[[email protected] ~]$
如果，这部分不是很熟悉，可以参考http://blog.csdn.net/liuge36/article/details/78592169
Step 3: Create a topic
[[email protected] ~]$ kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic flume_kafka
WARNING: Due to limitations in metric names, topics with a period (‘.‘) or underscore (‘_‘) could collide. To avoid issues it is best to use either, but not both.
Created topic "flume_kafka".
[[email protected] ~]$
Step 4: 开启之前的agent
[[email protected] conf]$ flume-ng agent --name a1 --conf . --conf-file ./test3.conf -Dflume.root.logger=INFO,console
Step 5: Start a consumer
kafka-console-consumer.sh --zookeeper hadoop000:2181 –topic flume-kafka上面的第五步执行之后，就会收到刷屏的结果，哈哈哈！！
上面的消费者会一直一直的刷屏，还是很有意思的!!!
这里的消费者是把接收到的数据数据到屏幕上
后面，我们会介绍，使用SparkStreaming作为消费者实时接收数据，并且接收到的数据做简单数据清洗的开发，从随机产生的日志中筛选出我们需要的数据.....
原文地址：https://www.cnblogs.com/liuge36/p/9883008.html