Spark Structured Streaming多Kafka主题异构消息模式实现问询

Awesome question! Moving from single-topic to multi-topic Kafka consumption with dynamic per-topic schemas in Spark Structured Streaming is a common scenario, and here's a practical, step-by-step approach that plays to Spark's strengths:

1. First, Read All Target Kafka Topics #

Start by configuring your Spark stream to consume all the topics you need. Spark supports comma-separated topic names or regex patterns for topic matching:

val rawKafkaStream = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker-1:9092,broker-2:9092")
  .option("subscribe", "app-topic-a,app-topic-b,app-topic-c") // Use subscribePattern for regex matching
  .load()

Keep an eye on the topic field in the raw stream—this will be your key to distinguishing data from different sources later.

2. Parse the Outer JSON Structure #

All your messages share a consistent outer structure (source, schema, message), so we can define a fixed schema to parse this layer first. We'll temporarily store the inner message as a key-value map since its structure varies per topic:

import org.apache.spark.sql.types._

// Fixed outer schema for all messages
val outerMessageSchema = StructType(Seq(
  StructField("source", StringType),
  StructField("schema", ArrayType(StructType(Seq(
    StructField("col_name", StringType),
    StructField("col_type", StringType)
  )))),
  StructField("message", MapType(StringType, StringType)) // Temp storage for dynamic content
))

val parsedStream = rawKafkaStream
  .selectExpr("CAST(value AS STRING) AS json_payload", "topic")
  .select(from_json($"json_payload", outerMessageSchema).alias("parsed_data"), $"topic")
  .select("parsed_data.source", "parsed_data.schema", "parsed_data.message", "topic")

3. Dynamically Apply Per-Topic Schemas to the Message Content #

This is the core of the solution. We'll convert the embedded schema field into a Spark StructType, then use it to transform the message map into a strongly-typed row. To optimize performance, we'll group messages by topic (assuming all messages in a topic share the same schema—adjust if your use case allows per-message schema changes).

First, create a helper function to convert the embedded schema array into a Spark-compatible schema:

def convertToSparkSchema(schemaRows: Seq[Row]): StructType = {
  val fields = schemaRows.map { row =>
    val colName = row.getAs[String]("col_name")
    val colDataType = row.getAs[String]("col_type") match {
      case "Integer" => IntegerType
      case "String" => StringType
      case "Double" => DoubleType
      case _ => StringType // Fallback for unrecognized types
    }
    StructField(colName, colDataType, nullable = true)
  }
  StructType(fields)
}

Next, use flatMapGroups to process messages per topic, applying the dynamic schema once per group:

val finalProcessedStream = parsedStream.groupBy("topic").flatMapGroups { (topic, messages) =>
  val messageList = messages.toList
  if (messageList.isEmpty) {
    Iterator.empty
  } else {
    // Extract the schema from the first message (assume topic schema consistency)
    val topicSchema = convertToSparkSchema(messageList.head.getAs[Seq[Row]]("schema"))
    // Add the original topic field to the final schema for traceability
    val finalSchema = topicSchema.add(StructField("source_topic", StringType))

    // Convert each message's map to a strongly-typed row
    messageList.map { msg =>
      val messageMap = msg.getAs[Map[String, String]]("message")
      // Map each field from the schema to the corresponding value in the message map
      val fieldValues = topicSchema.fields.map { field =>
        messageMap.get(field.name) match {
          case Some(value) => field.dataType match {
            case IntegerType => value.toInt
            case DoubleType => value.toDouble
            case StringType => value
            case _ => value
          }
          case None => null // Handle missing fields
        }
      }
      // Append the topic to the row
      Row.fromSeq(fieldValues :+ topic)
    }
  }
}(finalSchema) // Pass the final schema to Spark

4. Handle Schema Evolution (Optional) #

If your topics might evolve over time (e.g., adding new fields), you can enhance the convertToSparkSchema function to handle backward compatibility—for example, defaulting missing fields to null or using a merge strategy. For production systems, you might also want to offload schema storage to an external registry (like Confluent Schema Registry) instead of embedding schemas in messages, but this works for your current setup.

5. Write the Processed Stream to Your Target #

Finally, write the strongly-typed stream to your desired destination (Delta Lake, Parquet, another Kafka topic, etc.):

val writeQuery = finalProcessedStream.writeStream
  .format("delta")
  .option("path", "/your/storage/path/multi-topic-data")
  .option("checkpointLocation", "/your/checkpoint/path")
  .start()

writeQuery.awaitTermination()

Key Considerations #

• Schema Consistency: This approach assumes all messages in a topic share the same schema. If your use case requires per-message schema changes, skip grouping and process each message individually (note: this will impact performance).
• Error Handling: Add try-catch blocks in the type conversion logic to handle bad data (e.g., a non-integer string in an Integer field) gracefully.
• Performance: Grouping by topic reduces redundant schema parsing, which is critical for large-scale streams.

内容的提问来源于stack exchange，提问作者Kyle Schmitt