如何在Spark中实现多源流数据处理？含MySQL、HDFS、Kafka场景

一、Spark中多源流数据处理（HDFS静态数据 + Kafka流数据） #

Spark Structured Streaming是处理多源流场景的主流方案，核心是将静态数据（HDFS）和流数据（Kafka）统一为DataFrame/Dataset，通过关联、合并等操作完成整合。

常见整合模式 #

1. 流-静态数据关联：适用于需要用静态数据补充流数据维度信息的场景（比如流数据只有用户ID，静态数据存储用户详情）
2. 分别聚合后合并结果：适用于需要合并静态数据与流数据的统计指标的场景

代码示例 #

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

// 1. 读取HDFS静态结构化数据（以Parquet为例）
val hdfsStaticDF = spark.read.parquet("hdfs://cluster/path/to/static/data")

// 2. 读取并解析Kafka流数据
val kafkaStreamDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "kafka-broker-1:9092,kafka-broker-2:9092")
  .option("subscribe", "transaction-topic")
  .load()
  .selectExpr("CAST(value AS STRING) as raw_data")
  // 替换为你的数据Schema
  .select(from_json($"raw_data", StructType(Seq(
    StructField("user_id", StringType),
    StructField("amount", DoubleType),
    StructField("ts", TimestampType)
  ))).as("data"))
  .select("data.*")

// 模式1：流数据关联静态数据（基于user_id补充用户信息）
val joinedStream = kafkaStreamDF.join(hdfsStaticDF, Seq("user_id"), "left_outer")

// 模式2：分别聚合后合并指标
val staticAgg = hdfsStaticDF.groupBy("user_id").agg(sum("total_spend").as("history_spend"))
val streamAgg = kafkaStreamDF.groupBy("user_id").agg(sum("amount").as("current_spend"))
val combinedAgg = streamAgg.join(staticAgg, Seq("user_id"), "outer")
  .withColumn("total_spend", coalesce($"current_spend", lit(0)) + coalesce($"history_spend", lit(0)))

注意事项 #

• 若HDFS静态数据会定期更新，可通过foreachBatch在每个微批周期重新读取静态数据，或使用文件流监控HDFS目录的增量变化
• 关联操作需保证静态数据的分区合理，避免全表扫描导致性能瓶颈

二、业务场景落地：MySQL历史数据+Kafka新增数据近实时分析 #

针对5000万条MySQL交易数据导入HDFS、每日新增数据通过Kafka接入的场景，以下是端到端的实现方案：

1. 历史数据批量导入HDFS #

先将MySQL历史数据批量写入HDFS（推荐Parquet/ORC列式存储，压缩比高、查询性能优）：

val mysqlHistoryDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql://mysql-host:3306/transaction_db")
  .option("dbtable", "transactions")
  .option("user", "db_user")
  .option("password", "db_pass")
  .option("numPartitions", "20") // 并行读取，提升5000万条数据的导入速度
  .load()

// 按交易日期分区写入HDFS，方便后续增量查询
mysqlHistoryDF.write
  .partitionBy("transaction_date")
  .mode("overwrite")
  .parquet("hdfs://cluster/path/to/history/transactions")

可选：若需要定期同步MySQL增量历史数据，可基于transaction_id或update_time做增量拉取

2. 近实时多源数据整合分析 #

允许1-2分钟延迟，采用Structured Streaming微批模式（默认触发间隔为1分钟），核心思路是流数据聚合后与历史数据预聚合结果关联：

// 读取HDFS历史数据的预聚合结果（提前计算每日交易总额）
val historyDailyAgg = spark.read.parquet("hdfs://cluster/path/to/history/daily_agg")

// 读取Kafka新增交易流，按日聚合
val kafkaDailyStream = kafkaStreamDF
  .withColumn("transaction_date", to_date($"ts"))
  .groupBy("transaction_date")
  .agg(sum("amount").as("daily_stream_amount"))
  .withWatermark("transaction_date", "1 day") // 容忍1天内的迟到数据

// 合并历史与流数据的每日总额
val totalDailyAgg = kafkaDailyStream.join(
  historyDailyAgg,
  kafkaDailyStream("transaction_date") === historyDailyAgg("transaction_date"),
  "outer"
).select(
  coalesce(kafkaDailyStream("transaction_date"), historyDailyAgg("transaction_date")).as("transaction_date"),
  (coalesce($"daily_stream_amount", lit(0)) + coalesce($"daily_history_amount", lit(0))).as("total_amount")
)

3. 分析结果持久化 #

将结果保存到适合后续使用的存储系统：

方案A：写入HDFS供报表工具读取

totalDailyAgg.writeStream
  .format("parquet")
  .option("path", "hdfs://cluster/path/to/result/daily_transaction_agg")
  .option("checkpointLocation", "hdfs://cluster/path/to/checkpoint/daily_agg")
  .partitionBy("transaction_date")
  .trigger(Trigger.ProcessingTime("1 minute")) // 符合1-2分钟延迟要求
  .start()

方案B：写入MySQL供业务系统查询

totalDailyAgg.writeStream
  .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
    batchDF.write
      .format("jdbc")
      .option("url", "jdbc:mysql://mysql-host:3306/result_db")
      .option("dbtable", "daily_transaction_summary")
      .option("user", "db_user")
      .option("password", "db_pass")
      .mode("overwrite") // 或使用upsert逻辑更新现有日期的总额
      .save()
  }
  .trigger(Trigger.ProcessingTime("1 minute"))
  .start()

关键优化点 #

• 对历史数据提前做预聚合，避免流处理阶段关联全量历史数据
• 配置合理的并行度和分区数，适配集群资源
• 启用检查点机制，保证流任务故障后可以恢复

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