如何實現Spark Streaming和Kafka整合

本篇文章為大家展示了如何實現Spark Streaming和Kafka整合，內容簡明扼要并且容易理解，絕對能使你眼前一亮，通過這篇文章的詳細介紹希望你能有所收獲。

最近完成了Spark Streaming和Kafka的整合工作，耗時雖然不長，但是當中還是遇到了不少的坑，記錄下來，大家方便繞行。

先說一下環境：

Spark 2.0.0    kafka_2.11-0.10.0.0

之前的項目當中，已經在pom當中添加了需要的Spark Streaming的依賴，這次只需要添加Spark Streaming Kafka的以來就行了，問題來了。首先是我之前添加的Spark Streaming的依賴：

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming_2.11</artifactId>
      <version>2.0.0</version>
    </dependency>

然后是找到的spark streaming對kafka的支持依賴：

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka_2.11</artifactId>
    <version>1.6.2</version>
</dependency>

請注意2個version部分，好像差的有點多。不管了，照著例子寫寫看，果然報了各種class not found的錯誤。基本可以判斷是版本差異造成的問題。

可是，在http://mvnrepository.com上找不到更高版本的依賴怎么辦呢？

考慮了一下，只有一個辦法了，下載spark源碼，自行編譯打包需要的jar包。

在github上找到spark項目，clone下來，懶病又犯了，也沒仔細看當中的說明，直接就clean compile等等。結果又是各種報錯。好吧，好好看看吧，github上給了個地址：http://spark.apache.org/docs/latest/building-spark.html，照著做就沒問題了。

然后把項目當中pom里面對streaming kafka的依賴刪掉，引入我們自己生成的jar包：

spark-streaming-kafka-0-10_2.11-2.1.0-SNAPSHOT.jar

然后貼上代碼：

    val conf = new SparkConf().setAppName("kafkastream").setMaster("spark://master:7077").
      set("spark.driver.host", "192.168.1.142").
      setJars(List("/src/git/msgstream/out/artifacts/msgstream_jar/msgstream.jar",
        "/src/git/msgstream/lib/kafka-clients-0.10.0.0.jar",
        "/src/git/msgstream/lib/kafka_2.11-0.10.0.0.jar",
        "/src/git/msgstream/lib/spark-streaming-kafka-0-10_2.11-2.1.0-SNAPSHOT.jar"))
    val ssc = new StreamingContext(conf, Seconds(2))

    val topics = List("woozoom")
    val kafkaParams = Map(("bootstrap.servers", "master:9092,slave01:9092,slave02:9092"),
      ("group.id", "sparkstreaming"), ("key.deserializer", classOf[StringDeserializer]),
      ("value.deserializer", classOf[StringDeserializer]))
    val preferredHosts = LocationStrategies.PreferConsistent
    val offsets = Map(new TopicPartition("woozoom", 0) -> 2L)

    val lines = KafkaUtils.createDirectStream[String, String](
      ssc,
      preferredHosts,
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams, offsets))

    lines.foreachRDD(rdd => {
      rdd.foreach(x => {
        println(x)
      })
    })

    ssc.start()
    ssc.awaitTermination()

上面標紅的部分，是需要注意的，而這些本來我也是不會寫的，后來去到spark源碼找到test代碼

/src/git/spark/external/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/DirectKafkaStreamSuite.scala

測試，通過！！！

上述內容就是如何實現Spark Streaming和Kafka整合，你們學到知識或技能了嗎？如果還想學到更多技能或者豐富自己的知識儲備，歡迎關注億速云行業資訊頻道。