如何基于Spark实现MySQL两张动态增长表的持续关联？

Great question—this is a super common pain point when dealing with rapidly changing relational data and distributed processing. Let's break down why your current approaches fail, then walk through how CDC + Spark solves this reliably:

Why Your Current Methods Produce Inconsistent Results #

Before diving into solutions, let's clarify the root issues:

• Full-table Spark joins: Every time you run your job, you're loading a snapshot of the two tables. But since new rows are inserted every second, two runs will pull different snapshots—leading to different join results.
• MySQL-side joins + load: Even if you run the join directly in MySQL, unless you're using SERIALIZABLE isolation level (which kills performance), you'll hit phantom reads during the join. New rows inserted mid-query will be included in some runs but not others, making results inconsistent.

How CDC + Spark Fixes This #

CDC (Change Data Capture) lets you capture incremental changes (inserts, updates, deletes) from MySQL in real time, instead of relying on static snapshots. When paired with Spark's streaming capabilities, you can maintain a consistent, up-to-date view of both tables and perform joins that reflect all changes accurately.

Here's a step-by-step implementation:

1. Set Up CDC Capture for MySQL Tables #

First, you need to capture MySQL's data changes. The most popular open-source tool for this is Debezium:

• Enable MySQL's binlog (required for CDC) and set its format to ROW (Debezium needs row-level changes).
• Configure a Debezium connector to connect to your MySQL database. It will capture initial snapshots of both tables (to get your existing data) and then stream all subsequent changes to a message broker like Kafka.
• Each table's changes will go to a dedicated Kafka topic (e.g., mysql.your_db.table_a and mysql.your_db.table_b).

2. Process CDC Streams with Spark Structured Streaming #

Use Spark Structured Streaming to consume the Kafka topics, maintain state for both tables, and perform consistent joins:

Example Code (Scala)

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

// Define schemas matching your MySQL table structures
val tableASchema = StructType(Seq(
  StructField("id", IntegerType),
  StructField("common_id", IntegerType),
  StructField("data_a", StringType),
  StructField("ts", TimestampType)
))

val tableBSchema = StructType(Seq(
  StructField("id", IntegerType),
  StructField("common_id", IntegerType),
  StructField("data_b", StringType),
  StructField("ts", TimestampType)
))

// Read CDC stream for Table A from Kafka
val tableAStream = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "your-kafka-broker:9092")
  .option("subscribe", "mysql.your_db.table_a")
  .load()
  // Parse Debezium's JSON payload
  .select(from_json(col("value").cast("string"), tableASchema).as("payload"))
  .select("payload.*")
  // Add watermark to handle late data (adjust window based on your use case)
  .withWatermark("ts", "10 minutes")

// Read CDC stream for Table B from Kafka
val tableBStream = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "your-kafka-broker:9092")
  .option("subscribe", "mysql.your_db.table_b")
  .load()
  .select(from_json(col("value").cast("string"), tableBSchema).as("payload"))
  .select("payload.*")
  .withWatermark("ts", "10 minutes")

// Perform stream-stream join on the common field with time bounds
val consistentJoinedStream = tableAStream.join(
  tableBStream,
  (tableAStream("common_id") === tableBStream("common_id")) &&
  (tableAStream("ts") >= tableBStream("ts").minus(lit(10 minutes))) &&
  (tableAStream("ts") <= tableBStream("ts").plus(lit(10 minutes))),
  joinType = "inner"
)

// Write consistent results to target storage (e.g., Parquet, Iceberg)
val query = consistentJoinedStream.writeStream
  .format("parquet")
  .option("path", "/path/to/consistent-join-results")
  .option("checkpointLocation", "/path/to/spark-checkpoint") // Critical for fault tolerance
  .outputMode("append")
  .start()

query.awaitTermination()

3. Key Guarantees for Consistency #

• Exactly-Once Semantics: Spark's checkpointing ensures that if the job fails, it resumes from the last consistent state without reprocessing or losing data. Debezium + Kafka guarantees all changes are delivered at least once, and Spark's handling turns this into exactly-once.
• State Management: Spark maintains an in-memory (or disk-backed) state of both tables' latest records, so joins always use the most up-to-date data.
• Late Data Handling: The watermark configuration cleans up stale state and ensures you don't process outdated data, keeping your job efficient and results accurate.

4. Optional: Bootstrap with a Snapshot #

If you have a large amount of historical data, Debezium can capture an initial snapshot of both tables before streaming incremental changes. This lets you bootstrap your Spark state with existing data, so you don't have to reprocess years of changes from scratch.

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