Flume结合Spark测试

James

　　近日，在测试Flume结合Kafka结合Spark Streaming的实验。今天把Flume与Spark的简单结合做出来了，这里记录一下，避免网友走弯路。有不周到的地方还希望路过的大神多多指教。
　　

　　实验比较简单，分为两部分：一、使用avro-client发送数据 二、使用netcat发送数据
　　首先Spark程序需要Flume的两个jar包：

　　flume-ng-sdk-1.4.0、spark-streaming-flume_2.11-1.2.0
　　

　　一、使用avro-client发送数据
　　1、  编写Spark程序，该程序的功能是接收Flume事件
　　

　　
　　import org.apache.log4j.{Level, Logger}
　　import org.apache.spark.SparkConf
　　importorg.apache.spark.storage.StorageLevel
　　import org.apache.spark.streaming._
　　import org.apache.spark.streaming.flume._
　　object FlumeEventTest{
　　defmain(args:Array[String]) {
　　Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
　　Logger.getLogger("org.apache.eclipse.jetty.server").setLevel(Level.OFF)
　　val hostname = args(0)
　　val port = args(1).toInt
　　val batchInterval = args(2)
　　val sparkConf = newSparkConf().setAppName("FlumeEventCount").setMaster("local[2]")
　　val ssc = new StreamingContext(sparkConf,batchInterval)
　　valstream = FlumeUtils.createStream(ssc,hostname,port,StorageLevel.MEMORY_ONLY)
　　stream.count().map(cnt => "Received " + cnt + " flumeevents." ).print()
　　ssc.start()
　　ssc.awaitTermination()
　　}
　　}
　　

　　2、  Flume配置文件参数
　　

　　a1.channels = c1
　　a1.sinks = k1
　　a1.sources = r1
　　a1.sinks.k1.type = avro
　　a1.sinks.k1.channel = c1
　　a1.sinks.k1.hostname = localhost
　　a1.sinks.k1.port = 9999
　　a1.sources.r1.type = avro
　　a1.sources.r1.bind = localhost
　　a1.sources.r1.port = 44444
　　a1.sources.r1.channels = c1
　　a1.channels.c1.type = memory
　　a1.channels.c1.capacity = 1000
　　a1.channels.c1.transactionCapacity = 100
　　

　　这里，使用avro向flume的44444端口发送数据；然后flume通过9999向Spark发送数据。
　　

　　3、  运行Spark程序：

　　

　　4、  通过Flume配置文件启动Flumeagent
　　../bin/flume-ng agent --conf conf--conf-file ./flume-conf.conf --name a1
　　-Dflume.root.logger=INFO,console

　　Spark运行效果：

　　

　　5、  使用avro来发送文件：
　　./flume-ng avro-client --conf conf -Hlocalhost -p 44444 -F/opt/servicesClient/Spark/spark/conf/spark-env.sh.template-Dflume.root.logger=DEBUG,console
　　

　　Flume agent效果：

　　Spark效果：

　　

　　二、使用netcat发送数据
　　

　　1、  Spark程序同上
　　
　　2、  配置Flume参数
　　a1.channels = c1
　　a1.sinks = k1
　　a1.sources = r1
　　a1.sinks.k1.type = avro
　　a1.sinks.k1.channel = c1
　　a1.sinks.k1.hostname = localhost
　　a1.sinks.k1.port = 9999
　　a1.sources.r1.type = netcat
　　a1.sources.r1.bind = localhost
　　a1.sources.r1.port = 44444
　　a1.sources.r1.channels = c1
　　a1.channels.c1.type = memory
　　a1.channels.c1.capacity = 1000
　　a1.channels.c1.transactionCapacity = 100
　　

　　这里，使用telnet作为Flume的数据源
　　

　　3、  运行Spark程序同上
　　

　　4、  通过Flume配置文件启动Flumeagent
　　../bin/flume-ng agent --conf conf--conf-file ./flume-conf.conf --name a1
　　-Dflume.root.logger=INFO,console

　　注意：这里使用netcat作为Flume的数据源，注意与avro作为源的效果区别
　　

　　5、  使用telnet发送数据

　　

　　Spark效果：

　　

　　

　　这是两个比较简单的demo，如果真正在项目中使用Flume来收集数据，使用Kafka作为分布式消息队列，使用Spark Streaming实时计算，还需要详细研究Flume和Spark流计算。
　　

　　前段时间给部门做培训，演示了Spark Streaming的几个例子：文本处理、网络数据处理、stateful操作和window操作，这几天有时间整理整理，分享给大家。包括Spark MLlib的两个简单demo：基于K-Means的用户分类和基于协同过滤的电影推荐系统。
　　

　　

　　今天看了斯坦福Andrew Ng教授的ML课程，讲的很棒，这里把链接分享给大家：
　　http://open.163.com/special/opencourse/machinelearning.html