版权声明】博客内容由厦门大学数据库实验室拥有版权，未经允许，请勿转载！
返回Spark教程首页
推荐纸质教材：林子雨、郑海山、赖永炫编著《Spark编程基础（Python版）》

Flume是非常流行的日志采集系统，可以作为DStream的高级数据源。本部分将介绍如何让Flume推送消息给Spark Streaming，Spark Streaming收到消息后进行处理。

任务描述

把Flume Source设置为netcat类型，从终端上不断给Flume Source发送各种消息，Flume把消息汇集到Sink，这里把Sink类型设置为avro，由Sink把消息推送给Spark Streaming，由我们编写的Spark Streaming应用程序对消息进行处理。

Flume的安装和准备工作

下载安装Flume

请登录Linux系统（本教程全部统一采用hadoop用户登录），下载安装Flume。关于Flume的概念和安装方法，请参考厦门大学数据库实验室博客文章《日志采集工具Flume的安装与使用方法》，必须要理解什么是Flume Source、Flume Sink和Flume Agent等。这里我们从官网下载的安装文件是apache-flume-1.7.0-bin.tar.gz（官网下载地址）。这里假设读者已经按照博客文章成功安装了Flume，并且安装在“/usr/local/flume”目录下。

配置Flume数据源

请登录Linux系统，打开一个终端，执行如下命令新建一个Flume配置文件flume-to-spark.conf：

cd /usr/local/flume
cd conf
vim flume-to-spark.conf

在flume-to-spark.conf文件中写入如下内容：

#flume-to-spark.conf: A single-node Flume configuration
        # 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 = localhost
        a1.sources.r1.port = 33333

        # Describe the sink
        a1.sinks.k1.type = avro
        a1.sinks.k1.hostname = localhost
        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，绑定到localhost的33333端口，这样，我们后面就可以通过“telnet localhost 33333”命令向Flume Source发送消息。
同时，我们把Flume Sink类别设置为avro，绑定到localhost的44444端口，这样，Flume Source把采集到的消息汇集到Flume Sink以后，Sink会把消息推送给localhost的44444端口，而我们编写的Spark Streaming程序一直在监听localhost的44444端口，一旦有消息到达，就会被Spark Streaming应用程序取走进行处理。

特别要强调的是，上述配置文件完成以后，暂时“不要”启动Flume Agent，如果这个时候使用“flume-ng agent”命令启动agent，就会出现错误提示“localhost:44444拒绝连接”，也就是Flume Sink要发送消息给localhost的44444端口，但是，无法连接上localhost的44444端口。为什么会出现这个错误呢？因为，这个时候我们还没有启动Spark Streaming应用程序，也就没有启动localhost的44444端口，所以，Sink是无法向这个端口发送消息的。

Spark的准备工作

Kafka和Flume等高级输入源，需要依赖独立的库（jar文件）。按照我们前面安装好的Spark版本，这些jar包都不在里面。
现在我们就需要下载spark-streaming-flume_2.11-2.1.0.jar，其中2.11表示对应的Scala版本号，2.1.0表示Spark版本号。现在请在Linux系统中，打开一个火狐浏览器，请点击这里访问官网，里面有提供spark-streaming-flume_2.11-2.1.0.jar文件的下载。

下载后的文件会被默认保存在当前Linux登录用户的下载目录下，本教程统一使用hadoop用户名登录Linux系统，所以，文件下载后会被保存到“/home/hadoop/下载”目录下面。现在，我们在“/usr/local/spark/jars”目录下新建一个“flume”目录，就把这个文件复制到Spark目录的“/usr/local/spark/jars/flume”目录下。请新打开一个终端，输入下面命令：

cd /usr/local/spark/jars
mkdir flume
cd ~
cd 下载
cp ./spark-streaming-flume_2.11-2.1.0.jar /usr/local/spark/jars/flume

这样，我们就成功地把spark-streaming-flume_2.11-2.1.0.jar文件拷贝到了“/usr/local/spark/jars/flume”目录下。

下面我们还要修改spark目录下conf/spark-env.sh文件中的SPARK_DIST_CLASSPATH变量.把flume的相关jar包添加到此文件中

export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath):$(/usr/local/hbase/bin/hbase classpath):/usr/local/spark/examples/jars/*:/usr/local/spark/jars/kafka/*:/usr/local/kafka/libs/*:/usr/local/spark/jars/flume/*:/usr/local/flume/lib/*

这样，我们就已经准备好了Spark环境，它可以支持Flume相关编程了。

编写Spark程序使用Flume数据源

下面，我们就可以进行程序编写了。请新打开一个终端，然后，执行命令创建代码目录：

cd /usr/local/spark/mycode
mkdir flume
cd flume
vim FlumeEventCount.py

请在FlumeEventCount.py代码文件中输入以下代码：

from __future__ import print_function

import sys

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.flume import FlumeUtils
import pyspark
if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: flume_wordcount.py <hostname> <port>", file=sys.stderr)
        exit(-1)

    sc = SparkContext(appName="FlumeEventCount")
    ssc = StreamingContext(sc, 2)

    hostname= sys.argv[1]
    port = int(sys.argv[2])
    stream = FlumeUtils.createStream(ssc, hostname, port,pyspark.StorageLevel.MEMORY_AND_DISK_SER_2)
    stream.count().map(lambda cnt : "Recieve " + str(cnt) +" Flume events!!!!").pprint()

    ssc.start()
    ssc.awaitTermination()

测试程序效果

关闭之前打开的所有终端。首先，请新建第1个Linux终端，启动Spark Streaming应用程序，命令如下：

./bin/spark-submit --driver-class-path /usr/local/spark/jars/*:/usr/local/spark/jars/flume/* ./mycode/flume/FlumeEventCount.py localhost 44444

通过上面命令，我们为应用程序提供host和port两个参数的值分别为localhost和44444，程序会对localhost的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个终端不要关闭，让它一直处于监听状态。

现在，我们可以再另外新建第2个终端，在这个新的终端中启动Flume Agent，命令如下：

cd /usr/local/flume
bin/flume-ng agent --conf ./conf --conf-file ./conf/flume-to-spark.conf --name a1 -Dflume.root.logger=INFO,console

启动agent以后，该agent就会一直监听localhost的33333端口，这样，我们下面就可以通过“telnet localhost 33333”命令向Flume Source发送消息。第2个终端也不要关闭，让它一直处于监听状态。

请另外新建第3个终端，执行如下命令：

telnet localhost 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进行处理了。

至此，本实验顺利完成。实验结束后，要关闭各个终端，只要切换到该终端窗口，然后按键盘的Ctrl+C组合键，就可以结束程序运行。

厦大数据库实验室博客

Spark2.1.0+入门：把Flume作为DStream数据源(Python版)