Flume是非常流行的日志采集系统，可以作为DStream的高级数据源。本部分将介绍如何让Flume推送消息给Spark Streaming，Spark Streaming收到消息后进行处理。
任务描述
把Flume Source设置为netcat类型，从终端上不断给Flume Source发送各种消息，Flume把消息汇集到Sink，这里把Sink类型设置为avro，由Sink把消息推送给Spark Streaming，由我们编写的Spark Streaming应用程序对消息进行处理。
配置Flume数据源
在flume-to-spark.conf文件中写入如下内容：
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = master
a1.sources.r1.port = 33333# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = master
a1.sinks.k1.port =44444# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000000
a1.channels.c1.transactionCapacity = 1000000# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
在上面的配置文件中，我们把Flume Source类别设置为netcat，绑定到master的33333端口，这样，我们后面就可以通过“telnet master 33333”命令向Flume Source发送消息。
同时，我们把Flume Sink类别设置为avro，绑定到 master 的44444端口，这样，Flume Source把采集到的消息汇集到Flume Sink以后，Sink会把消息推送给 master 的44444端口，而我们编写的Spark Streaming程序一直在监听 master 的44444端口，一旦有消息到达，就会被Spark Streaming应用程序取走进行处理。
特别要强调的是，上述配置文件完成以后，暂时“不要”启动Flume Agent，如果这个时候使用“flume-ng agent”命令启动agent，就会出现错误提示“master:44444拒绝连接”，也就是Flume Sink要发送消息给 master 的44444端口，但是，无法连接上 master 的44444端口。为什么会出现这个错误呢？因为，这个时候我们还没有启动Spark Streaming应用程序，也就没有启动 master 的44444端口，所以，Sink是无法向这个端口发送消息的。
编写Spark程序使用Flume数据源
FlumeEventCount.scala
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming._
import org.apache.spark.streaming.flume._object FlumeEventCount {def main(args: Array[String]) {if (args.length < 2) {System.err.println("Usage: FlumeEventCount <host> <port>")System.exit(1)}val Array(host, port) = argsval batchInterval = Milliseconds(2000)// Create the context and set the batch sizeval sparkConf = new SparkConf().setAppName("FlumeEventCount").setMaster("local[2]")val ssc = new StreamingContext(sparkConf, batchInterval)// Create a flume streamval stream = FlumeUtils.createStream(ssc, host, port.toInt, StorageLevel.MEMORY_ONLY_SER_2)// Print out the count of events received from this server in each batchstream.count().map(cnt => "Received " + cnt + " flume events." ).print()ssc.start()ssc.awaitTermination()}
}
FlumeEventCount.scala程序在编译后运行时，需要我们提供host和port两个参数，程序会对指定的host和指定的port进行监听，Milliseconds(2000)设置了时间间隔为2秒，所以，该程序每隔2秒就会从指定的端口中获取由Flume Sink发给该端口的消息，然后进行处理，对消息进行统计，打印出“Received 0 flume events.”这样的信息。
built.sbt
name := "StreamingFlume"version := "0.1"scalaVersion := "2.11.8"libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"
libraryDependencies += "org.apache.spark" % "spark-streaming_2.11" % "2.1.0"
libraryDependencies += "org.apache.spark" % "spark-streaming-flume_2.11" % "2.1.0"
测试程序效果
首先，请新建第1个Linux终端，启动Spark Streaming应用程序，命令如下：
[root@master ~]# spark2-submit --class FlumeEventCount streamingflume_2.11-0.1.jar master 44444
通过上面命令，我们为应用程序提供host和port两个参数的值分别为 master 和44444，程序会对 master 的44444端口进行监听，Milliseconds(2000)设置了时间间隔为2秒，所以，该程序每隔2秒就会从指定的端口中获取由Flume Sink发给该端口的消息，然后进行处理，对消息进行统计，打印出“Received 0 flume events.”这样的信息。
执行该命令后，屏幕上会显示程序运行的相关信息，并会每隔2秒钟刷新一次信息，大量信息中会包含如下重要信息：
-------------------------------------------
Time: 1488029430000 ms
-------------------------------------------
Received 0 flume events.
因为目前Flume还没有启动，没有给FlumeEventCount发送任何消息，所以Flume Events的数量是0。
第1个终端不要关闭，让它一直处于监听状态。
现在，我们可以在 Cloudera Manager 中启动 Flume Agent。
启动agent以后，该agent就会一直监听 master 的 33333 端口，这样，我们下面就可以通过“telnet master 33333”命令向Flume Source发送消息。
请另外新建一个个终端，执行如下命令：
[root@master ~]# telnet master 33333
执行该命令以后，就可以在这个窗口里面随便敲入若干个字符和若干个回车，这些消息都会被Flume监听到，Flume把消息采集到以后汇集到Sink，然后由Sink发送给Spark的FlumeEventCount程序进行处理。然后，你就可以在运行FlumeEventCount的前面那个终端窗口内看到类似如下的统计结果：
-------------------------------------------
Time: 1488029430000 ms
-------------------------------------------
Received 0 flume events.
-------------------------------------------
Time: 1488029432000 ms
-------------------------------------------
Received 8 flume events.
-------------------------------------------
Time: 1488029434000 ms
-------------------------------------------
Received 21 flume events.
从屏幕信息中可以看出，我们在telnet那个终端内发送的消息，都被成功发送到Spark进行处理了。