Я пытаюсь собрать потребителя kafka в Scala, используя Intellij для чтения сообщений из topi kafka c. У меня есть spark и kafka на windows.
Я пробовал этот код :
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka010._
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
object connector {
def main(args: Array[String]) {
class Kafkaconsumer {
val kafkaParams = Map[String, Object](
"bootstrap.servers" -> "host1:port,host2:port2,host3:port3",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> "use_a_separate_group_id_for_each_stream",
"auto.offset.reset" -> "latest",
"enable.auto.commit" -> (false: java.lang.Boolean)
)
val sparkConf = new SparkConf().setMaster("yarn")
.setAppName("kafka example")
val streamingContext = new StreamingContext(sparkConf, Seconds(10))
val topics = Array("topicname")
val topicsSet = topics.split(",").toSet
val stream = KafkaUtils.createDirectStream[String, String](
streamingContext,PreferConsistent,Subscribe[String, String](kafkaParams,topicsSet)
)
stream.print()
stream.map(record => (record.key, record.value))
streamingContext.start()
streamingContext.awaitTermination()
}
}
Но у меня есть ошибка в этих двух строках
val topicsSet = topics.split(",").toSet
streamingContext,PreferConsistent,Subscribe[String, String](kafkaParams,topicsSet)
Функция разделения и функция подписки всегда отображаются красным.
Есть идеи ?
Спасибо