1 概述

Apache Flume是一種分散式，可靠且可用的服務，用於高效地收集，彙總和移動大量日誌資料。 這裡我們學習如何配置Flume和Spark Streaming來接收來自Flume的資料。 提供兩種方法來解決這問題。

注意：從Spark 2.3.0開始，不推薦使用Flume支援。個人也不推薦這種架構，資料量小的情況下可能沒什麼問題，但是再資料量過大的情況下Streaming流式處理是處理不過來的，必定會造成訊息堵塞，所以建議加一個訊息中介軟體例如Kafka。

2 Push-based

方法1：流水線式推送方式
Flume接收的資料直接交給Spark Streaming處理。 所以這種方法中，Spark Streaming需要一個接收器receiver，作為Flume的Avro代理，Flume可以將資料推送到該接收器。 所以這裡要注意local的使用。

啟動要求：先啟動Spark Streaming進行接收資料再啟動Flume.
選型結構：netcat => memory ==> avro sink ==> streaming

• 編寫程式碼

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

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

    //傳入的引數hostname,port
    val Array(hostname,port)=args
      val sparkConf=new SparkConf().setAppName
 
("FlumePush").setMaster("local[2]")
      val ssc=new StreamingContext(sparkConf,Seconds(10))

    // import FlumeUtils
    import org.apache.spark.streaming.flume._
    //create input DStream as follows.
    val flumeStream=FlumeUtils.createStream(ssc,hostname,port.toInt)

    //詞頻統計
    flumeStream.map(x
 
=>new String(x.event.getBody.array()).trim)
      .flatMap(x=>x.split(",")).map((_,1)).reduceByKey(_+_).print()

    ssc.start()
    ssc.awaitTermination()
  }

}

上面我們是通過外面進行傳遞hostname和port，所以要進行下面的配置

在Programe arguments傳遞引數。作者使用的時windows本地執行出現各種錯誤，選擇打包上傳就不需要在本地配置引數的，直接在spark-submit的時候傳遞引數。

• 配置檔案

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

# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = localhost
a1.sinks.k1.port = 41414

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity=10000
a1.channels.c1.transactionCapacity=1000

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

啟動命令

./spark-submit --master local[2] \
--class cn.zhangyu.FlumePush \
--name FlumePush \
/home/hadoop/lib/spark_streaming-1.0-SNAPSHOT.jar \
localhost 41414

這時候提交應用程式的時候肯定會報錯，因為我們本地並沒有FlumeUtils這個jar包，所以要指定--packages g:a:v引數(有網路才可以用，沒有網路可以實現下載好通過–jars引數指定)

./spark-submit --master local[2] \
--class cn.zhangyu.FlumePush \
--packages org.apache.spark:spark-streaming-flume_2.11:2.2.0 \
--name FlumePush \
/home/hadoop/lib/spark_streaming-1.0-SNAPSHOT.jar \
localhost 41414

./flume-ng agent \
--name a1 \
--conf $FLUME_HOME/conf \
--conf-file $FLUME_HOME/conf/flume-sparkStreaming-push \
-Dflume.root.logger=INFO,console

輸入資料：

[[email protected] conf]$ telnet localhost 44444
Trying ::1...
Connected to localhost.
Escape character is '^]'.
a,a,a,b,b,b,c,c,c,d,d,s
OK
a,a,a,b,b,b,c,c,c,d,d,s
OK

結果：

(d,2)
(b,3)
(s,1)
(a,3)
(c,3)

(d,2)
(b,3)
(s,1)
(a,3)
(c,3)

3 Pull-based

使用該種方式Flume不是直接將資料推送到Spark Streaming，而是執行一個自定義的Flume接收器，它允許執行以下操作。

1. Flume將資料推入接收器，並且資料保持緩衝。
2. Spark Streaming使用可靠的Flume接收器和並從接收器中提取資料。該事務只有在Spark Streaming接收和複製資料後才會成功。

注意：
1.這時候並不需要Spark Streaming端的receiver了。
2.啟動順序:先啟動flume在啟動spark streaming。

這確保了比以前的方法更強大的可靠性和容錯保證。 但是，這需要配置Flume執行自定義接收器。 以下是配置步驟。

• flume配置
  1.jar包依賴

groupId = org.apache.spark
artifactId = spark-streaming-flume-sink_2.11
version = 2.3.0

 groupId = org.scala-lang
 artifactId = scala-library
 version = 2.11.8

 groupId = org.apache.commons
 artifactId = commons-lang3
 version = 3.5

2.配置檔案

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

# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 44444

# Describe the sink
a1.sinks.k1.type = org.apache.spark.streaming.flume.sink.SparkSink
a1.sinks.k1.hostname = localhost
a1.sinks.k1.port = 41414

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity=10000
a1.channels.c1.transactionCapacity=1000

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

• 啟動

./flume-ng agent \
--name a1 \
--conf $FLUME_HOME/conf \
--conf-file $FLUME_HOME/conf/flume-sparkStreaming-poll \
-Dflume.root.logger=INFO,console

./spark-submit --master local[2] \
--class cn.zhangyu.FlumePull \
--packages org.apache.spark:spark-streaming-flume_2.11:2.2.0 \
--name FlumePull \
/home/hadoop/lib/spark_streaming-1.0-SNAPSHOT.jar \
localhost 41414

• 輸入

telnet localhost 44444

a,a,a,b,b,b,c,c,c,d,d,s

• 結果：

(d,2)
(b,3)
(s,1)
(a,3)
(c,3)

4 maven中scope屬性的

擴充套件：上面我們說了使用spark-submit提交任務的時候可以使用--packagers,--jars但是這並不是一個很好的方法，我們可以使用maven編譯的時候加上需要的類，做法如下：
1.在pom.xml新增

<plugin>
        <artifactId>maven-assembly-plugin</artifactId>
        <configuration>
          <archive>
            <manifest>
              <mainClass></mainClass>
            </manifest>
          </archive>
          <descriptorRefs>
            <descriptorRef>jar-with-dependencies</descriptorRef>
          </descriptorRefs>
        </configuration>
      </plugin>

Dependency Scope

在POM 4中，dependency中還引入了scope，它主要管理依賴的部署。目前可以使用5個值：

• compile，預設值，適用於所有階段，會隨著專案一起釋出。
• provided，類似compile，期望JDK、容器或使用者會提供這個依賴。如servlet.jar。
• runtime，只在執行時使用，如JDBC驅動，適用執行和測試階段。
• test，只在測試時使用，用於編譯和執行測試程式碼。不會隨專案釋出。
• system，類似provided，需要顯式提供包含依賴的jar，Maven不會在Repository中查詢它。

依賴範圍控制哪些依賴在哪些classpath 中可用，哪些依賴包含在一個應用中。讓我們詳細看一下每一種範圍：

所以我們可以在編譯的時候把FlumeUtils這個類加進去：

使用mvn assembly:assembly命令進行編譯