Spark Streaming（二）集成Flume数据

Spark Streaming集成Flume有两种方式，分别是基于Push的和基于Pull的，本篇文档参考Spark官网，基于Spark 2.2.0和Flume 1.6.0

• Push-based

这种方式是Flume通过Agent去push数据，Spark Streaming使用一个receiver接收数据，这种基于push的方式，必须先启动Spark Streaming的receiver接收数据，然后再启动Flume

1. 编写Flume Agent:flume_push_streaming.conf，Agent采用netcat为source，avro的sink以及memery的channel:

simple-agent.sources = netcat-source
simple-agent.sinks = avro-sink
simple-agent.channels = memory-channel

simple-agent.sources.netcat-source.type = netcat
simple-agent.sources.netcat-source.bind = hadoop000
simple-agent.sources.netcat-source.port = 44444

simple-agent.sinks.avro-sink.type = avro
simple-agent.sinks.avro-sink.hostname = 192.168.199.203
simple-agent.sinks.avro-sink.port = 41414

simple-agent.channels.memory-channel.type = memory

simple-agent.sources.netcat-source.channels = memory-channel
simple-agent.sinks.avro-sink.channel = memory-channel

2. 编写Spark Streaming应用程序

引入如下spark streaming整合flume的依赖：

<dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming-flume_2.11</artifactId>
      <version>${spark.version}</version>
</dependency>

应用程序核心代码如下：

if(args.length!=2){
      System.err.print("Usage:FlumePushWordCount <hostname> <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))
val flumeStream = FlumeUtils.createStream(ssc, hostname, port.toInt)
flumeStream.map(x=>new String(x.event.getBody.array()).trim())
      .flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()
ssc.start()
ssc.awaitTermination()

3. 生产环境运行
   因为Spark Streaming集成Flume的jar包并未打到程序包里，所以spark-submit启动的时候需要通过--packages org.apache.spark:spark-streaming-flume_2.11:2.2.0添加该jar包，第一次会先去下载jar包，速度会稍慢，第二次就可以直接使用了，详细命令如下：

spark-submit \
--class com.yxzc.FlumePushWordCount \
--master local[2] \
--packages org.apache.spark:spark-streaming-flume_2.11:2.2.0 \
/home/hadoop/lib/sparktrain-1.0.jar \
hadoop000 41414

如果生产环境不能连接外网，或者网速很差时，可以先从maven仓库下载该jar包，然后在spark-submit是通过--jars指定该jar，个人比较推荐这种方式

4. 启动Flume
   步骤1里已经编写好了Flume Agent，这些直接启动就可以了，启动命令：

flume-ng agent  \
--name simple-agent   \
--conf $FLUME_HOME/conf    \
--conf-file $FLUME_HOME/conf/flume_push_streaming.conf  \
-Dflume.root.logger=INFO,console

5. telnet（telnet hadoop000 44444）输入数据，查看Spark Streaming程序输出结果

• Pull-based
  这种方式是Flume推送数据到一个sink里缓存起来，然后Spark Streaming程序从该sink拉取数据，这种方式成功的前提是数据收到了并被Spark Streaming以多副本的方式成功保存数据，这种方式比push更可靠，容错性更高，这种方式应该先启动Flume，然后再启动Spark Streaming程序

1. 编写Flume Agent:flume_pull_streaming.conf,Agent采用netcat的source,SparkSink和memery的channel

simple-agent.sources = netcat-source
simple-agent.sinks = spark-sink
simple-agent.channels = memory-channel

simple-agent.sources.netcat-source.type = netcat
simple-agent.sources.netcat-source.bind = hadoop000
simple-agent.sources.netcat-source.port = 44444

simple-agent.sinks.spark-sink.type = org.apache.spark.streaming.flume.sink.SparkSink
simple-agent.sinks.spark-sink.hostname = hadoop000
simple-agent.sinks.spark-sink.port = 41414

simple-agent.channels.memory-channel.type = memory

simple-agent.sources.netcat-source.channels = memory-channel
simple-agent.sinks.spark-sink.channel = memory-channel

2. 编写Spark Streaming应用程序
   引入如下spark streaming整合flume的依赖：

<dependency>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-streaming-flume-sink_2.11</artifactId>
     <version>${spark.version}</version>
</dependency>
<dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-library</artifactId>
    <version>${scala.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-lang3</artifactId>
    <version>3.5</version>
</dependency>

应用程序核心代码如下：

if(args.length!=2){
    System.err.print("Usage:FlumePullWordCount <hostname> <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))
val flumeStream = FlumeUtils.createPollingStream(ssc, hostname, port.toInt)
flumeStream.map(x=>new String(x.event.getBody.array()).trim())
      .flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()
    
ssc.start()
ssc.awaitTermination()

3. 启动Flume
   步骤1里已经编写好了Flume Agent，这些直接启动就可以了，启动命令：

flume-ng agent  \
--name simple-agent   \
--conf $FLUME_HOME/conf    \
--conf-file $FLUME_HOME/conf/flume_pull_streaming.conf  \
-Dflume.root.logger=INFO,console

4. 生产环境运行
   这里跟push方式没有区别，可以参考上边push方式，详细启动命令为：

spark-submit \
--class com.yxzc.FlumePullWordCount \
--master local[2] \
--packages org.apache.spark:spark-streaming-flume_2.11:2.2.0 \
/home/hadoop/lib/sparktrain-1.0.jar \
hadoop000 41414

5. telnet（telnet hadoop000 44444）输入数据，查看Spark Streaming程序输出结果

Spark Streaming作为一个流处理框架，Flume作为一个日志收集框架，直接对接其实是有问题的，尤其不同时间段产生日志的数量差异较大时，负载均衡、吞吐量都对机器性能有所要求和限制，所以一般使用Kafka作为一个缓冲队列，从Flume过来的数据先缓存到Kafka里，Spark Streaming从Kafka里获取数据。