環境版本：Scala 2.10.5; Spark 1.6.0; Kafka 0.10.0.1; Flume 1.6.0

Flume/Kafka的安裝配置請看我之前的部落格:
http://blog.csdn.net/dr_guo/article/details/52130812
http://blog.csdn.net/dr_guo/article/details/51050715

1.Flume 配置檔案 kafka_sink.conf（監聽檔案1.log sink到kafka）

agent.sources = r1
agent.channels = c2
agent.sinks = k2

####define source begin

##define source-r1-exec
agent.sources.r1.channels = c2
agent.sources.r1.type = exec
agent.sources.r1.command = tail -F /home/drguo/flumetest/1.log


####define sink begin

##define sink-k2-kafka
agent.sinks.k2.channel = c2
agent.sinks.k2.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.k2.topic = test
agent.sinks.k2.brokerList = DXHY-YFEB-01:9092,DXHY-YFEB-02:9092,DXHY-YFEB-03:9092
agent.sinks.k2.requiredAcks = 1
agent.sinks.k2.batchSize = 20
#agent.sinks.k2.serializer.class = Kafka.serializer.StringEncoder
agent.sinks.k2.producer.type = async
#agent.sinks.k2.batchSize = 100


####define channel begin

##define channel-c2-memory
agent.channels.c2.type = memory
agent.channels.c2.capacity = 1000000
agent.channels.c2.transactionCapacity = 100
 

2.啟動Flume
flume-ng agent --conf conf --conf-file ./kafka_sink.conf --name agent -Dflume.root.logger=INFO,console  

3.執行Spark Streaming(word count)

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

/**
  * 單詞統計 | Spark Streaming 接收並處理 Kafka 中資料
  * Created by drguo on 2017/11/21.
  */
object KafkaWordCount {
  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[2]")

    val ssc = new StreamingContext(sparkConf, Seconds(50))//方塊長度
    //ssc.checkpoint("checkpoint")

    val topics = "test"
    val numThreads = "2"
    val zkQuorum = "DXHY-YFEB-01:2181,DXHY-YFEB-02:2181,DXHY-YFEB-03:2181"
    val group = "TestConsumerGroup"

    val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
    // Create a direct stream
    val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)

    val words = lines.flatMap(_.split(" "))
    /**
     reduceByKeyAndWindow(_ + _, _ - _, Seconds(15), Seconds(10), 2)
     總的來說SparkStreaming提供這個方法主要是出於效率考慮。 
     比如每10秒計算一下前15秒的內容，（每個batch 5秒）， 
那麼就是通過如下步驟:
    1.儲存上一個window的reduce值
    2.計算出上一個window的begin 時間到 重複段的開始時間的reduce 值 =》 oldRDD
    3.重複時間段的值結束時間到當前window的結束時間的值 =》 newRDD
    4.重複時間段的值等於上一個window的值減去oldRDD

    這樣就不需要去計算每個batch的值， 只需加加減減就能得到新的reduce出來的值。

  */
    val wordCounts = words.map((_, 1)).reduceByKeyAndWindow(
      (v1: Int, v2: Int) => {
        v1 + v2
      },
      //十分鐘 視窗長度
      //Minutes(10),
      //每五秒 視窗移動長度 統計十分鐘內的資料
      Seconds(50))

    wordCounts.print()

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

4.插入測試資料
[[email protected] flumetest]# echo "hadoop spark hello world world hh hh" >> 1.log

5.檢視執行結果

Flume簡單的命令列操作
#建立topic
[[email protected] KAFKA]# bin/kafka-topics --create --zookeeper DXHY-YFEB-01:2181,DXHY-YFEB-02:2181,DXHY-YFEB-03:2181 --replication-factor 3 -partitions 1 --topic test

#檢視topic
[
 
[email protected] KAFKA]# bin/kafka-topics --list --zookeeper DXHY-YFEB-01:2181,DXHY-YFEB-02:2181,DXHY-YFEB-03:2181

#使用生產者向topic生產訊息 可以有多個生產者
[[email protected] KAFKA]# bin/kafka-console-producer --broker-list DXHY-YFEB-01:9092,DXHY-YFEB-02:9092,DXHY-YFEB-03:9092 --topic test
[[email protected] KAFKA]# ~
[[email protected] KAFKA]# ~

#使用消費者消費topic裡的訊息 可以有多個消費者
[[email protected] KAFKA]# bin/kafka-console-consumer --zookeeper DXHY-YFEB-01:2181,DXHY-YFEB-02:2181,DXHY-YFEB-03:2181 --from-beginning --topic test  
[[email protected] KAFKA]# ~ 
[[email protected] KAFKA]# ~

#檢視消費者組
[[email protected] KAFKA]# bin/kafka-consumer-groups --bootstrap-server DXHY-YFEB-01:9092,DXHY-YFEB-02:9092,DXHY-YFEB-03:9092 --list