Flume结合Spark测试-创新互联

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

创新互联长期为上1000家客户提供的网站建设服务，团队从业经验10年，关注不同地域、不同群体，并针对不同对象提供差异化的产品和服务；打造开放共赢平台，与合作伙伴共同营造健康的互联网生态环境。为云城企业提供专业的成都网站建设、网站制作，云城网站改版等技术服务。拥有十余年丰富建站经验和众多成功案例,为您定制开发。

实验比较简单，分为两部分：一、使用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

另外有需要云服务器可以了解下创新互联scvps.cn，海内外云服务器15元起步，三天无理由+7*72小时售后在线，公司持有idc许可证，提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案，具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势，专为企业上云打造定制，能够满足用户丰富、多元化的应用场景需求。

网页题目：Flume结合Spark测试-创新互联
文章网址：http://cxhlcq.com/article/dcopoo.html