scala spark-streaming整合kafka （spark 2.3 kafka 0.10）

?Maven组件如下：? ?

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
    <version>2.3.0</version>
</dependency>

?官网代码如下：

?pasting

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License,Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

// scalastyle:off println
package org.apache.spark.examples.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka010._

/**
 * Consumes messages from one or more topics in Kafka and does wordcount.
 * Usage: DirectKafkaWordCount <brokers> <topics>
 *   <brokers> is a list of one or more Kafka brokers
 *   <topics> is a list of one or more kafka topics to consume from
 *
 * Example:
 *    $ bin/run-example streaming.DirectKafkaWordCount broker1-host:port,broker2-host:port 
 *    topic1,topic2
 */
object DirectKafkaWordCount {
def main(args: Array[String]) {
if (args.length < 2) {
      System.err.println(s"""
        |Usage: DirectKafkaWordCount <brokers> <topics>
        |  <brokers> is a list of one or more Kafka brokers
        |  <topics> is a list of one or more kafka topics to consume from
        |
        """.stripMargin)
      System.exit(1)
    }

    StreamingExamples.setStreamingLogLevels()

val Array(brokers,topics) = args

// Create context with 2 second batch interval
    val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount")
val ssc = new StreamingContext(sparkConf,Seconds(2))

// Create direct kafka stream with brokers and topics
    val topicsSet = topics.split(",").toSet
val kafkaParams = Map[String,String]("metadata.broker.list" -> brokers)
val messages = KafkaUtils.createDirectStream[String,String](
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String,String](topicsSet,kafkaParams))

// Get the lines,split them into words,count the words and print
    val lines = messages.map(_.value)
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x,1L)).reduceByKey(_ + _)
    wordCounts.print()

// Start the computation
    ssc.start()
    ssc.awaitTermination()
  }
}
// scalastyle:on println

?

运行以上代码出现如下错误等：

?Exception in thread "main" org.apache.kafka.common.config.ConfigException: Missing required configuration "bootstrap.servers" which has no default value.

解决方法：

? 由错误可见，是因为没有设置kafka相关参数。

?把官网代码修改如下：

package cn.xdf.userprofile.stream
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds,StreamingContext}
import org.apache.spark.streaming.kafka010._

import scala.collection.mutable

object DirectKafka {
def main(args: Array[String]): Unit = {
if (args.length < 2) {
      System.err.println(
s"""
           |Usage: DirectKafkaWordCount <brokers> <topics>
           |  <brokers> is a list of one or more Kafka brokers
           |  <topics> is a list of one or more kafka topics to consume from
           |
        """.stripMargin)
      System.exit(1)
    }
val Array(brokers,topics)=args

var conf = new SparkConf()
        .setAppName("DirectKafka")
          .setMaster("local[2]")

val ssc = new StreamingContext(conf,Seconds(2))

val topicsSet=topics.split(",").toSet
val kafkaParams=mutable.HashMap[String,String]()
//必须添加以下参数，否则会报错
         kafkaParams.put("bootstrap.servers" ,brokers)
          kafkaParams.put("group.id","group1")
         kafkaParams.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
         kafkaParams.put("value.deserializer" ,"org.apache.kafka.common.serialization.StringDeserializer")
val messages=KafkaUtils.createDirectStream [String,String](
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String,String](topicsSet,kafkaParams
          )
      )
// Get the lines,count the words and print
      val lines = messages.map(_.value)
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x,1L)).reduceByKey(_ + _)
      wordCounts.print()

// Start the computation
      ssc.start()
      ssc.awaitTermination()

  }
}

?运行如下：

?

? ?bin/kafka-server-start ./etc/kafka/server.properties &

?

?

运行spark

?/usr/local/spark-2.3.0/bin/spark-submit --class cn.xdf.userprofile.stream.DirectKafka --master yarn --driver-memory 2g ? ? --num-executors 1 ? ? ?--executor-memory 2g ? ? --executor-cores 1 ?userprofile2.0.jar localhost:9092 test?

?

?启动生产者

?

?

> ?hello me

?

查看结果：