Scala에서 Kafka 소비자를 컴파일 할 수 없습니다.

Question

Scala 2.10.6 및 Spark 1.6.2을 사용하여 카프카 항목의 메시지를 사용하고 싶습니다. 이 코드는 잘 컴파일

<dependency> 
    <groupId>org.apache.spark</groupId> 
    <artifactId>spark-streaming_2.10</artifactId> 
    <version>1.6.2</version> 
</dependency> 

그러나 나는 auto.offset.reset을 정의하려면 여기에 문제가 발생 : 내가 kafkaParams를 추가 할 때

val topicMap = topic.split(",").map((_, kafkaNumThreads.toInt)).toMap 
val data = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap, 
          StorageLevel.MEMORY_AND_DISK_SER_2).map(_._2) 

것은, 그것은 더 이상 컴파일되지 않습니다 카프카를 위해 나는이 종속성을 사용하고 있습니다 :

val kafkaParams = Map[String, String](
"zookeeper.connect" -> zkQuorum, "group.id" -> group, 
"zookeeper.connection.timeout.ms" -> "10000", 
"auto.offset.reset" -> "smallest") 

val data = KafkaUtils.createStream(ssc, kafkaParams, topicMap, 
           StorageLevel.MEMORY_AND_DISK_SER_2).map(_._2) 

오류 메시지 :

94: error: missing parameter type for expanded function ((x$3) => x$3._2) 
[ERROR]             StorageLevel.MEMORY_AND_DISK_SER_2).map(_._2) 

createStream의 매개 변수가 여러 가지로 시도되었지만 모두 실패합니다. 누군가 제발 도와 드릴까요?

Answer 1

KafkaUtils.createStream에 유형 매개 변수를 추가하여 스트림의 기본 유형을 해결해야합니다. 예를 들어, 키와 값이 String :

val data: DStream[String] = 
    KafkaUtils 
    .createStream[String, String, StringDecoder, StringDecoder](
     ssc, 
     kafkaParams, 
     topicMap, 
     StorageLevel.MEMORY_AND_DISK_SER_2 
).map(_._2)