Я пытаюсь прочитать данные от потребителя kafka с помощью spark2-shell.
Пожалуйста, найдите мой код ниже.
Я запускаю свою оболочку spark2 следующим образом:
spark2-shell --jars kafka-clients-0.10.1.2.6.2.0-205.jar, spark-sql-kafka-0-10_2.11-2.1.1.jar
И, пожалуйста, найдите мой код ниже:
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming._
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import spark.implicits._
val ssc = new StreamingContext(sc, Seconds(2))
val topics = List("testingtopic01")
val preferredHosts = LocationStrategies.PreferConsistent
val kafkaParams = Map(
"bootstrap.servers" -> "localhost:9192",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"security.protocol" -> "SASL_PLAINTEXT",
"auto.offset.reset" -> "earliest",
"group.id" -> "spark-streaming-consumer-group"
)
val lines = KafkaUtils.createDirectStream[String, String](
ssc,
preferredHosts,
ConsumerStrategies.Subscribe[String, String](topics.distinct, kafkaParams)
)
lines.print()
ssc.start()Но после того, как я запускаю поток искры, здесь ничего не появляется.
scala> ssc.start()
18/12/19 15:50:07 WARN streaming.StreamingContext:DynamicAllocation is enabled for this application.Enabling Dynamic allocation for Spark Streaming applications can cause data loss if Write Ahead Log is not enabled for non-replayable sources like Flume. See the programming guide for details on how to enable the Write Ahead Log.
Пожалуйста, предложите мне способ обойти эту проблему.
Заранее спасибо.
