Spark Streaming 两种方式连接 Flume

绝地灬酷狼 2022-12-05 01:37 196阅读 0赞

Spark Streaming 通过 **Push** 和 **Pull** 两种方式对接 Flume 数据源。以 Spark Streaming 的角度来看，Push 方式属于推送（由 Flume 向 Spark 推送）；而 Pull 属于拉取（Spark 拉取 Flume 的输出）。

不论以何种方式，开发过程类似，都是由 Spark Streaming 对接 Flume 数据流，Flume 做为 Spark Streaming 的数据源。Push 和 Pull 两者的差别主要体现在Flume Sink 的不同，而 Flume Source 与 Channel 不会受影响。在演示示例时，Flume Source 以 nectcat 为例，Channel 为 memory，重点关注 Sink 的变化。在下文中也是如此。

添加额外依赖包，spark的依赖包我就不介绍了。

<dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-streaming-flume_2.11</artifactId>
         <version>2.2.0</version>
    </dependency>

## Push 方式 ##

当采用 Push 方式时，Flume 架构为：netcat->memory->avro。

这里的 avro sink 需要指定 avro 服务地址，容易混淆的是 avro 服务由 Flume还是 Spark 启动问题。要搞清楚这个问题，首先看 Push 方式在 Spark 端如何实现。如下代码所示。

object FlumePushWordCount { 
     def main(args: Array[String]) { 
       if (args.length < 2) { 
          System.err.println(
             "Usage: FlumePushWordCount <host> <port>")
          System.exit(1)
       }
    
    
       val Array(hostname, port) = args
       val sparkConf = new SparkConf()
                     .setAppName("FlumePushWordCount").setMaster("local[2]")
       val ssc = new StreamingContext(sparkConf, Seconds(5))
    
    
    
       //推送方式：flume 向 spark 发送数据
       val flumeStream = FlumeUtils.createStream(ssc, hostname, port.toInt, 
                                        StorageLevel.MEMORY_ONLY_SER_2)
    
    
    
       //flume 中的数据通过 event.getBody()才能拿到真正的内容
       flumeStream.map(x =>new String(x.event.getBody.array()).trim)
          .flatMap(_.split(" "))
          .map(word => (word, 1)) // 每个单词映射成一个 pair
          .reduceByKey(_ + _) // 根据每个 key 进行累加
          .print() // 打印前 10 个数据
       ssc.start()
       ssc.awaitTermination()
    
      } 
    
    
    }

跟踪 FlumeUtils.createStream()源码如下图所示:

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center]

FlumeUtils.createStream()会创建一个 FlumeReceiver 用于接收 Flume 的数据。实际上是创建了一个 avro 服务（org.apache.avro.ipc.NettyServer）。

现在可以得出结论，该 avro 服务应该由 Spark 启动，所以在 Flume集成Spark，  
在 Pull 方式下对接运行时，必须先启动 Spark 流应用程序。否则报错如下图所示：

![在这里插入图片描述][20200908192930508.png_pic_center]

对于主要代码的编写比较简单，相对麻烦的是测试代码的过程。归纳测试运行过程步骤如下：

1.  编写 Pull 方式获取 Inut DSt代码（如上述代码）并打包。
2.  编写 Flume 配置文件。如下面 flume\_push\_streaming.conf 所示：

agent.sources = s1 
    agent.channels = c1 
    agent.sinks = sk0 sk1 
    
    
    #设置 Source 的类型为 netcat，使用的 channel 为 c1 
    agent.sources.s1.type = netcat 
    agent.sources.s1.bind = localhost 
    agent.sources.s1.port = 44444 
    agent.sources.s1.channels = c1 
    
    
    #设置 Sink 为 logger 模式，使用的 channel 为 c1 
    agent.sinks.sk0.type=logger
    agent.sinks.sk0.channel=c1
    
    
    #AvroSink 向 Spark(41414)推送数据
    agent.sinks.sk1.type=avro
    agent.sinks.sk1.hostname=hadoop101
    agent.sinks.sk1.port=5555
    agent.sinks.sk1.channel = c1 
    
    
    #设置 channel 信息 
    #内存模式
    agent.channels.c1.type = memory 
    agent.channels.c1.capacity = 1000

1.  启动 Spark 流应用程序。如下面 spark-submit.flume\_push\_streaming 所示：  
    其中 flume-ng-sdk 为 Flume 自带组件，spark-streaming-flume 为 Flume 与 SparkStreaming 集成组件，scala-demo1-1.0-SNAPSHOT.jar 是第一步的打包结果，需要 提前准备这些jar包，并根据实际情况调整下面代码中的目录信息。

# spark-submit.flume_push_streaming
    
    
    spark-submit --class cn.comp.spark.streaming.FlumePushWordCount \
    
    --jars ./jars/flume-ng-sdk-1.6.0-cdh5.5.0.jar, \
    
    jars/spark-streaming-flume_2.11-2.2.0.jar \
    
    scala-demo1-1.0-SNAPSHOT.jar localhost 41414

1.  启动 Flume Agent。如下面 flume\_push\_streaming-start.sh 所示：

#flume_push_streaming-start.sh 视实际 Flume 安装目录修改
    
    
    flume-ng agent \
    --name agent \
    
    --conf-file ./flume_push_streaming.conf \
    
    -Dflume.root.logger=INFO,console

1.  启动 nc 或者 Telnet 连接 Socket，向 Flume nectcat Source 发送数据，并在Spark 流应用程序中观察结果。如下二图所示：

![在这里插入图片描述][20200908194050707.png_pic_center]

![在这里插入图片描述][20200908194127918.png_pic_center]

## Pull 方式 ##

当采用 Pull 方式时，Flume 架构为：netcat->memory->SparkSink。  
此 Sink 全类名为 **org.apache.spark.streaming.flume.sink.SparkSink**，需要在Flume “.conf”文件中使用，该类包含在 spark-streaming-flume-sink 组件中。所以还需要将此组件放在$FLUME\_HOME/lib 下。具体源码如下图所示：  
![在这里插入图片描述][20200908194320530.png_pic_center]

SparkSink 由 Flume 启动 avro 服务，所以本质上仍然是 avro Sink，如下图所示：  
然后由 Spark 连接后轮循数据，意味着 Flume Agent 应该先启动。  
![在这里插入图片描述][2020090819453476.png_pic_center]

由 Flume 启动 SparkSink，如下图所示：  
![在这里插入图片描述][20200908194713421.png_pic_center]

这也是为什么在使用 Pull 方式时，Flume 需要 spark-streaming-flume-sink 组件的原因。  
该Sink继承了Flume的抽象类AbstractSink，意味着在使用spark-submit提交 Spark 流应用程序时也需要 flume-ng-sdk 组件。如下面组件依赖层次图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 1]

具体运行步骤如下：

1.编写 Pull 方式获取 Input DStream 代码并打包。

object FlumePullWordCount { 
     def main(args: Array[String]) { 
       if (args.length < 2) { 
          System.err.println(
              "Usage: FlumePushWordCount <host> <port>")
          System.exit(1)
        }
    
     
       val Array(hostname, port) = args
       val sparkConf = new SparkConf()
                       .setAppName("FlumePullWordCount").setMaster("local[2]")
       val ssc = new StreamingContext(sparkConf, Seconds(5))
    
    
       //获取 flume 数据
       val flumeStream = FlumeUtils.createPollingStream(ssc, 
                                  hostname,port.toInt,StorageLevel.MEMORY_ONLY_SER_2)
    
    
       flumeStream.map(x => new String(x.event.getBody.array()).trim)
            .flatMap(_.split(" "))
            .map(word => (word, 1)) // 每个单词映射成一个 pair
           .reduceByKey(_ + _) // 根据每个 key 进行累加
           .print() // 打印前 10 个数据
       ssc.start()
       ssc.awaitTermination()
    
      } 
    
    }

2.编写 Flume 配置文件。如下面 streaming\_pull\_flume.conf 所示：

#streaming_pull_flume.conf
    agent.sources = s1 
    agent.channels = c1 
    agent.sinks = sk1 
    
    
    #设置 Source 的内省为 netcat，使用的 channel 为 c1 
    agent.sources.s1.type = netcat 
    agent.sources.s1.bind = localhost 
    agent.sources.s1.port = 44444 
    agent.sources.s1.channels = c1 
    
    
    #设置 Sink 为 logger 模式，使用的 channel 为 c1 
    agent.sinks.sk0.type=logger
    agent.sinks.sk0.channel=c1
    
    
    #SparkSink,要求 flume lib 目录存在 spark-streaming-flume-sink_2.11-x.x.x.jar
    agent.sinks.sk1.type=org.apache.spark.streaming.flume.sink.SparkSink
    agent.sinks.sk1.hostname=localhost
    agent.sinks.sk1.port=41414
    agent.sinks.sk1.channel = c1 
    
    
    #设置 channel 信息 
    #内存模式
    agent.channels.c1.type = memory 
    agent.channels.c1.capacity = 1000

3.启动 Flume Agent。如下面 streaming\_pull\_flume-start.sh 所示：

flume-ng agent \
    
    --name agent \
    --conf-file ./streaming_pull_flume.conf \ 
    
    -Dflume.root.logger=INFO,console

4.启动 Spark 流应用程序。如下面 spark-submit.streaming\_pull\_flume 所示：

spark-submit --class cn.comp.spark.streaming.FlumePullWordCount \
    --jars ./jars/flume-ng-sdk-1.6.0-cdh5.5.0.jar,\
    
    
    jars/spark-streaming-flume_2.11-2.2.0.jar,\
    jars/spark-streaming-flume-sink_2.11-2.2.0.jar \
    
    scala-demo1-1.0-SNAPSHOT.jar localhost 41414

5.启动 nc 或者 Telnet 连接 Socket，向 Flume nectcat Source 发送数据，并在Spark 流应用程序中观察结果。  
如下二图所示：

![在这里插入图片描述][20200908195651103.png_pic_center]

Spark 流应用程序运行结果：  
![在这里插入图片描述][20200908195744188.png_pic_center]

## 总结 ##

两种方式进行对比如下：

1.  对第三方 jar 的依赖：Pull 方式时 Flume 需要调用 Spark 实现的 Sink，所 以多依赖一个spark-streaming-flume-sink，Flume 与 Spark 两边同时依赖。
2.  启动顺序：Push 方式时 Flume 虽然不依赖 Spark，但 avro sink 需要一个 avro服务，这个服务应该先启动。所以 Spark 应先启动。而 Pull 则与之相反。用一个比喻形容便是：谁主动干活就可以不用启动 avro 服务，并且 avro 服务必须先运行。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/a1f3fb4b464246349c3ec91a50fbea3b.png
[20200908192930508.png_pic_center]: /images/20221123/b016cf63c1f14c899e38ac2a0ed472da.png
[20200908194050707.png_pic_center]: /images/20221123/3f6f0c511f8a44ef9e85fdf2e9771227.png
[20200908194127918.png_pic_center]: /images/20221123/1a3aa4a530a24d6e99abdcf14fb1f178.png
[20200908194320530.png_pic_center]: /images/20221123/19a173072f854338bc88390afb6204ed.png
[2020090819453476.png_pic_center]: /images/20221123/c0bd1cbcb79d4d5cbb3dfcbceb82c261.png
[20200908194713421.png_pic_center]: /images/20221123/5d75b77151944077844c9d9d0356aec0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/5d7b2b61e2b74024bfc88220e5762dae.png
[20200908195651103.png_pic_center]: /images/20221123/502942a0d8d04045907c5e91559aa9aa.png
[20200908195744188.png_pic_center]: /images/20221123/8b0fee65f0b3483f9e8cf8fb840aaaaf.png