1

2

3

4

5

<dependency>

<groupId>org.apache.spark</groupId>

<artifactId

<version>2.0.2</version>

</dependency>

01

02

03

04

05

06

07

08

09

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

package cn.itcast.dstream.kafka

import org.apache.spark.{SparkConf, SparkContext}

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

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

import org.apache.spark.streaming.kafka.KafkaUtils

import scala.collection.immutable

//todo:利用sparkStreaming對接kafka實現單詞計數----採用receiver(高階API)

object SparkStreamingKafka_Receiver {

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

//1、建立sparkConf

val sparkConf: SparkConf = new SparkConf()

.setAppName("SparkStreamingKafka_Receiver")

.setMaster("local[4]")

.set("spark.streaming.receiver.writeAheadLog.enable","true") //開啟wal預寫日誌，儲存資料來源的可靠性

//2、建立sparkContext

val sc = new SparkContext(sparkConf)

sc.setLogLevel("WARN")

//3、建立StreamingContext

val ssc = new StreamingContext(sc,Seconds(5))

//設定checkpoint

ssc.checkpoint("./Kafka_Receiver")

//4、定義zk地址

val zkQuorum="node1:2181,node2:2181,node3:2181"

//5、定義消費者組

val groupId="spark_receiver1"

//6、定義topic相關資訊 Map[String, Int]

// 這裡的value並不是topic分割槽數，它表示的topic中每一個分割槽被N個執行緒消費

val topics=Map("spark_kafka" -> 2)

//7、通過KafkaUtils.createStream對接kafka

//這個時候相當於同時開啟3個receiver接受資料

val receiverDstream: immutable.IndexedSeq[ReceiverInputDStream[(String, String)]] = (1 to 3).map(x => {

val stream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc, zkQuorum, groupId, topics)

stream

}

)

//使用ssc.union方法合併所有的receiver中的資料

val unionDStream: DStream[(String, String)] = ssc.union(receiverDstream)

//8、獲取topic中的資料

val topicData: DStream[String] = unionDStream.map(_._2)

//9、切分每一行,每個單詞計為1

val wordAndOne: DStream[(String, Int)] = topicData.flatMap(_.split(" ")).map((_,1))

//10、相同單詞出現的次數累加

val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_+_)

//11、列印輸出

result.print()

//開啟計算

ssc.start()

ssc.awaitTermination()

}

}

01

02

03

04

05

06

07

08

09

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

package cn.itcast.dstream.kafka

import kafka.serializer.StringDecoder

import org.apache.spark.{SparkConf, SparkContext}

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

import org.apache.spark.streaming.dstream.{DStream, InputDStream}

import org.apache.spark.streaming.kafka.KafkaUtils

//todo:利用sparkStreaming對接kafka實現單詞計數----採用Direct(低階API)

object SparkStreamingKafka_Direct {

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

//1、建立sparkConf

val sparkConf: SparkConf = new SparkConf()

.setAppName("SparkStreamingKafka_Direct")

.setMaster("local[2]")

//2、建立sparkContext

val sc = new SparkContext(sparkConf)

sc.setLogLevel("WARN")

//3、建立StreamingContext

val ssc = new StreamingContext(sc,Seconds(5))

ssc.checkpoint("./Kafka_Direct")

//4、配置kafka相關引數

val kafkaParams=Map("metadata.broker.list"->"node1:9092,node2:9092,node3:9092","group.id"->"Kafka_Direct")

//5、定義topic

val topics=Set("spark01")

//6、通過 KafkaUtils.createDirectStream接受kafka資料，這裡採用是kafka低階api偏移量不受zk管理

val dstream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,topics)

//7、獲取kafka中topic中的資料

val topicData: DStream[String] = dstream.map(_._2)

//8、切分每一行,每個單詞計為1

val wordAndOne: DStream[(String, Int)] = topicData.flatMap(_.split(" ")).map((_,1))

//9、相同單詞出現的次數累加

val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_+_)

//10、列印輸出

result.print()

//開啟計算

ssc.start()

ssc.awaitTermination()

}

}