Spark-Streaming整合Kafka實作wordcount

配置版本資訊：spark-2.3.4，Kafka-2.10，Scala-2.11，JDK8

1.建立Maven工程

配置Pom檔案

<properties>
        <spark.version>2.3.4</spark.version>
        <kafka.version>2.1.0</kafka.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>${kafka.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
    </dependencies>
           

連接配接Kafka并進行詞頻統計

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}

object SSKafka_Direct {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("streaming")
    val ssc = new StreamingContext(sparkConf,Duration(10000))  //采集周期10s
    //TODO SparkStreaming讀取Kafka的資料
    //kafka配置資訊
    val kafkaPara: Map[String, Object] = Map[String, Object](
      //zookeeper位址	
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "slaver1:9092,slaver2:9092,slaver3:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "app",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )

    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] =
      KafkaUtils.createDirectStream[String, String](
        ssc,
        LocationStrategies.PreferConsistent,
        //訂閱的topic名kafka_spark
        ConsumerStrategies.Subscribe[String, String](Set("kafka_spark"), kafkaPara))
    val valueDStream: DStream[String] = kafkaDStream.map(record=>record.value())
    valueDStream.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()

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

注1：new不出Scala檔案

1.檢查是否安裝了Scala插件

2.檢查是否将目錄設定為source

3.檢查是否導入scala JDK，在projectStructure—>Modules—>Dependencies中+—>Libraray—>scalaSDK

注2：在運作的時候報錯：找不到或者加載不到主類

版本不相容，使用上文提供的POM檔案即可解決

2.建立生産者，發送資料

依次啟動zookeeper，Hadoop，Kafka，spark

建立topic，生産者

kafka-topics.sh --create --zookeeper slaver1:2181 --replication-factor 1 --partitions 3 --topic kafka_spark
kafka-console-producer.sh --broker-list slaver1:9092 --topic  kafka_spark
           

啟動程式，在消費者端口輸入單詞，運作結果如下：