Спарк Streaming Кафка потребитель не нравится DStream

Question

Я использую Спарк Shell (Scala 2,10 и Спарк Streaming org.apache.spark:spark-streaming-kafka-0-10_2.10:2.0.1), чтобы проверить искру/Кафка потребитель:

import org.apache.kafka.clients.consumer.ConsumerRecord 
import org.apache.kafka.common.serialization.StringDeserializer 
import org.apache.spark.streaming.kafka010._ 
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent 
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe 
import org.apache.spark._ 
import org.apache.spark.streaming._ 
import org.apache.spark.streaming.dstream.DStream 

val kafkaParams = Map[String, Object](
    "bootstrap.servers" -> "mykafka01.example.com:9092", 
    "key.deserializer" -> classOf[StringDeserializer], 
    "value.deserializer" -> classOf[StringDeserializer], 
    "group.id" -> "mykafka", 
    "auto.offset.reset" -> "latest", 
    "enable.auto.commit" -> (false: java.lang.Boolean) 
) 

val topics = Array("mytopic") 

def createKafkaStream(ssc: StreamingContext, topics: Array[String], kafkaParams: Map[String,Object]) : DStream[(String, String)] = { 
    KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams)) 
} 

def messageConsumer(): StreamingContext = { 
    val ssc = new StreamingContext(SparkContext.getOrCreate(), Seconds(10)) 

    createKafkaStream(ssc, topics, kafkaParams).foreachRDD(rdd => { 
     rdd.collect().foreach { msg => 
      try { 
       println("Received message: " + msg._2) 
      } catch { 
       case e @ (_: Exception | _: Error | _: Throwable) => { 
       println("Exception: " + e.getMessage) 
       e.printStackTrace() 
      } 
      } 
     } 
    }) 

    ssc 
} 

val ssc = StreamingContext.getActiveOrCreate(messageConsumer) 
ssc.start() 
ssc.awaitTermination() 

Когда я запускаю это я получаю следующее исключение:

<console>:60: error: type mismatch; 
found : org.apache.spark.streaming.dstream.InputDStream[org.apache.kafka.clients.consumer.ConsumerRecord[String,String]] 
required: org.apache.spark.streaming.dstream.DStream[(String, String)] 
        KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams)) 
                  ^

Я проверил Scala/API Docs снова и снова, и этот код выглядит как он должен выполнить правильно. Любая идея, где я собираюсь сходить?

Answer 1

Subscribe принимает topics аргумент как Array[String], вы передаете одну строку согласно def createKafkaStream(ssc: StreamingContext, topics: String,. Изменение типа аргумента на Array[String] (и его правильное обращение) устранит проблему.