具有独特消息模式的Spark结构化流媒体多个Kafka主题

Question

{
  "source": "Application A",
  "schema": [{"col_name": "countryId", "col_type": "Integer"}, {"col_name": "name", "col_type": "String"}],
  "message": {"countryId": "21", "name": "Poland"}
}

Answer 1

当用你的kafka生产者发送这些事件时，你也可以发送一个键和值。如果每个事件都有它的事件类型作为键，那么当从主题读取流时，您还可以获得键：

val kafkaKvPair = spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .load()
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)] 

然后，您只需筛选要处理的事件：

val events = kafkaKvPair
  .filter(f => f._1 == "MY_EVENT_TYPE")

这样，如果您在一个Spark应用程序中订阅了多个主题，您可以处理任意多个事件类型。