如何在每批 Spark Streaming 作业中使用不同的 Kafka 主题？

Question

我很确定没有简单的方法可以做到这一点，但这是我的用例：

我有一个 Spark Streaming 作业（版本 2.1.0），每个微批次的持续时间为 5 秒。

我的目标是在每个微批处理间隔使用 1 个不同主题的数据，总共 250 个 Kafka 主题。你可以把下面的代码作为一个简单的例子：

 val groupId:String = "first_group"
 val kafka_servers:String =  "datanode1:9092,datanode2:9092,datanode3:9092"

 val ss:SparkSession = SparkSession.builder().config("spark.streaming.unpersist","true").appName("ConsumerStream_test").getOrCreate()
 val ssc:StreamingContext= new StreamingContext(ss.sparkContext,Duration(5000))

val kafka_parameters:Map[String,Object]=Map(
"bootstrap.servers"       -> kafka_servers,
"key.deserializer"        -> classOf[StringDeserializer],
"value.deserializer"      -> classOf[ByteArrayDeserializer],
"heartbeat.interval.ms"   -> (1000:Integer),
"max.poll.interval.ms"    -> (100:Integer),
"enable.auto.commit"      -> (false: java.lang.Boolean),
"autoOffsetReset"         -> OffsetResetStrategy.EARLIEST,
//"connections.max.idle.ms" -> (5000:Integer),
"group.id"                -> groupId
)

val r = scala.util.Random
val kafka_list_one_topic=List("topic_"+ r.nextInt(250))

val consumer:DStream[ConsumerRecord[String,Array[Byte]]] = KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferBrokers, ConsumerStrategies.
Subscribe[String, Array[Byte]](kafka_list_one_topic , kafka_parameters))

consumer.foreachRDD( eachRDD => {
     // DOING SOMETHING WITH THE DATA...
  })
ssc.start()
ssc.awaitTermination()

但是这种方法的问题是，Spark 只会运行一次初始代码（foreachRDD 命令之前的所有内容），以便创建 Kafka 消费者 DStream，但在接下来的微批处理中，它只运行“foreachRDD "声明。

例如，假设 r.nextInt(250) 返回 40。Spark Streaming 作业将连接到 topic_40 并处理其数据。但是在接下来的微批处理中，它仍然会连接到topic_40，并且忽略foreachRDD语句之前的所有命令。

我猜这是意料之中的，因为 foreachRDD 语句之前的代码只在 Spark 驱动程序上运行。

我的问题是，有没有一种方法可以做到这一点，而不必每 5 秒重新启动一次 Spark 应用程序？

谢谢。

Answer 1

我的方法很简单，如果你希望它是真正随机的并且不关心任何其他后果，请将 kafka_list_one_topic 设为可变变量并在流代码中更改它。

val r = scala.util.Random
var kafka_list_one_topic=List("topic_"+ r.nextInt(250))

val consumer:DStream[ConsumerRecord[String,Array[Byte]]] = 
KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferBrokers, 
ConsumerStrategies.
Subscribe[String, Array[Byte]](kafka_list_one_topic , kafka_parameters))

consumer.foreachRDD( eachRDD => {
 // DOING SOMETHING WITH THE DATA...
 kafka_list_one_topic=List("topic_"+ r.nextInt(250))
 })
ssc.start()
ssc.awaitTermination()