Flume Push

<!-- Spark Streaming 整合 Flume -->
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-flume_2.11</artifactId>
    <version>${spark.version}</version>
</dependency>

本地测试

$ vim  flume-push-streaming.conf
simple-agent.sources = netcat-source
simple-agent.channels = memory-channel
simple-agent.sinks = avro-sink

simple-agent.sources.netcat-source.type = netcat
simple-agent.sources.netcat-source.bind = host000
simple-agent.sources.netcat-source.port = 9999

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

simple-agent.sinks.avro-sink.type = avro
simple-agent.sinks.avro-sink.hostname = 192.168.1.100 
simple-agent.sinks.avro-sink.port = 10000

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




import org.apache.spark.SparkConf
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * push方式
  */
object FlumePushWordCount {
  def main(args: Array[String]): Unit = {
    if (args.length != 2) { // 写活，从运行参数中读取hostname和port
      System.err.println("Usage: FlumePushWordCount <hostname> <port>")
      System.exit(1)
    }

    val conf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("FlumePushWordCount")
      .set("spark.driver.host", "localhost")
    val ssc = new StreamingContext(conf, Seconds(5))

    // 从Flume接收数据
    val flumeStream = FlumeUtils.createStream(ssc, args(0), args(1).toInt)
    flumeStream.map(x => new String(x.event.getBody.array()).trim)
      .flatMap(_.split(" "))
      .map((_,1))
      .reduceByKey(_+_)
      .print()

    ssc.start()
    ssc.awaitTermination()
  }
}

1. 运行本地代码，运行时参数加入0.0.0.0 10000
2. $ flume-ng agent \
 --name simple-agent \
 --conf $FLUME_HOME/conf \
 --conf-file /home/user000/confs/flume_conf/flume-push-streaming.conf \
 -Dflume.root.logger=INFO,console
3. $ telnet host000 9999

服务器测试

修改代码
//      .setMaster("local[2]")
//      .setAppName("FlumePushWordCount")
//      .set("spark.driver.host", "localhost")
$ mvn clean package -DskipTests
$ scp spark-learning-1.0-SNAPSHOT.jar user000@host000:~/jars

修改flume配置
simple-agent.sinks.avro-sink.hostname = host000

运行
// 有--packages需要联网
$ spark-submit  --master local[2] \
--class FlumePushWordCount \
--packages org.apache.spark:spark-streaming-flume_2.11:2.1.0 \
~/jars/spark-learning-1.0-SNAPSHOT.jar host000 10000

$ flume-ng agent \
 --name simple-agent \
 --conf $FLUME_HOME/conf \
 --conf-file /home/user000/confs/flume_conf/flume-push-streaming.conf \
 -Dflume.root.logger=INFO,console

$ telnet host000 9999

Flume Pull

<!-- Scala -->
<dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-library</artifactId>
    <version>${scala.version}</version>
</dependency>

<!-- Spark Streaming整合Flume Pull方式 -->
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-flume-sink_2.11</artifactId>
    <version>${spark.version}</version>
</dependency>

<!-- 依赖的工具包 -->
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-lang3</artifactId>
    <version>3.5</version>
</dependency>

本地测试

$ vim flume_pull_streaming.conf
simple-agent.sources = netcat-source
simple-agent.channels = memory-channel
simple-agent.sinks = spark-sink

simple-agent.sources.netcat-source.type = netcat
simple-agent.sources.netcat-source.bind = host000
simple-agent.sources.netcat-source.port = 9999

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

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

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


import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.flume.FlumeUtils

/**
  * pull
  */
object FlumePullWordCount {
  def main(args: Array[String]): Unit = {
    if (args.length != 2) {
      System.err.println("Usage: FlumePushWordCount <hostname> <port>")
      System.exit(1)
    }

    val conf = new SparkConf()
          .setMaster("local[2]")
          .setAppName("FlumePullWordCount")
          .set("spark.driver.host", "localhost")
    val ssc = new StreamingContext(conf, Seconds(5))

    // 与Push只有 FlumeUtils.createPollingStream 这一点不同，然后就变成了主动pull数据
    val flumeStream = FlumeUtils.createPollingStream(ssc, args(0), args(1).toInt)
    flumeStream.map(x => new String(x.event.getBody.array()).trim)
      .flatMap(_.split(" "))
      .map((_,1))
      .reduceByKey(_+_)
      .print()

    ssc.start()
    ssc.awaitTermination()
  }
}


运行
$ flume-ng agent \
--name simple-agent \
--conf $FLUME_HOME/conf \
--conf-file /home/user000/confs/flume_conf/flume_pull_streaming.conf \
-Dflume.root.logger=INFO,console

$ 运行本地代码，加入参数host000 10000
$ telnet host000 9999

服务器测试

修改代码
//          .setMaster("local[2]")
//          .setAppName("FlumePullWordCount")
//          .set("spark.driver.host", "localhost")


$ mvn clean package -DskipTests
$ scp spark-learning-1.0-SNAPSHOT.jar user000@host000:~/jars

运行
$ flume-ng agent \
 --name simple-agent \
 --conf $FLUME_HOME/conf \
 --conf-file /home/user000/confs/flume_conf/flume_pull_streaming.conf \
 -Dflume.root.logger=INFO,console

$ spark-submit  --master local[2] \
--class FlumePullWordCount \
--packages org.apache.spark:spark-streaming-flume_2.11:2.1.0 \
~/jars/spark-learning-1.0-SNAPSHOT.jar host000 10000

$ telnet host000 9999