1.需求：
SparkStreaming處理來自flume push方式發(fā)來的數(shù)據(jù),即flume將數(shù)據(jù)推給spark Streaming

2.代碼：
(1)pom.xml

<dependencies>
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.1.0</version>

        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-flume -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-flume_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>

    </dependencies>

(2)flume文件option
option

#bin/flume-ng agent -n a1 -f myconf/option -c conf -Dflume.root.logger=INFO,console
#定義agent名，source寄悯，channel，sink的名稱
a1.sources=r1
a1.channels =c1
a1.sinks=k1

#具體定義source
a1.sources.r1.type= spooldir
a1.sources.r1.spoolDir= /opt/TestFolder/logs

#具體定義channel1
a1.channels.c1.type = memory
a1.channels.c1.capacity=10000
a1.channels.c1.transactionCapacity = 100

#具體定義sink
a1.sinks = k1
a1.sinks.k1.type = avro
a1.sinks.k1.channels =c1
a1.sinks.k1.hostname=192.168.31.211
a1.sinks.k1.port=1236

#組裝source, channel,sink

a1.sources.r1.channels = c1
a1.sinks.k1.channel =c1

(3)

package day1211
import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.flume.FlumeUtils

object MyFlumeStream {

  def main(args: Array[String]): Unit = {

    System.setProperty("hadoop.home.dir", "/Users/macbook/Documents/hadoop/hadoop-2.8.4")
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    val conf = new SparkConf().setAppName("MyFlumeStream").setMaster("local[2]")

    val ssc = new StreamingContext(conf,Seconds(3))

    //創(chuàng)建 flume event 從 flume中接收push來的數(shù)據(jù) ---> 也是DStream
    //flume將數(shù)據(jù)push到了 ip 和 端口中
    val flumeEventDstream = FlumeUtils.createStream(ssc, "192.168.1.121", 1236)

    val lineDStream = flumeEventDstream.map( e => {
      new String(e.event.getBody.array)
    })

    lineDStream.print()

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

3.運行：
(1)運行SparkStreaming程序：
(2)開啟flume

bin/flume-ng agent -n a1 -c conf -f jobconf/option -Dflume.root.logger=INFO,console

4.向/opt/TestFolder/logs中添加數(shù)據(jù)，查看結(jié)果：

cp a.txt /opt/TestFolder/logs

添加數(shù)據(jù)

5.結(jié)果

結(jié)果