Spark Streaming 如何实现不同DataFrame并行写入多张表

完全可以实现多个DataFrame的并行写入，核心思路是并行提交Spark Action任务，避免串行触发作业，具体方案及注意事项如下：

前置优化：避免重复计算 #

由于两个子DataFrame都是从同一个Kafka源DF转换拆分得到，需要先对父DF做缓存，防止两次写入作业重复拉取、计算Kafka源数据：

# 缓存Kafka读取转换后的父DF，可根据内存情况选择存储级别
parent_df.persist(StorageLevel.MEMORY_AND_DISK_SER)
# 触发Action让缓存实际落地
parent_df.count()

# 你的拆分逻辑，得到两个子DF
attr1_df =  # 处理attribute1Formatted字段得到的DF
attr2_df =  # 处理attribute2Formatted字段得到的DF

方案1：批处理场景用多线程并行提交写任务 #

Spark支持多线程提交作业，调度器会并行执行多个写入任务，PySpark示例如下：

from concurrent.futures import ThreadPoolExecutor

# 通用写表函数
def write_table(df: DataFrame, table_name: str):
    df.write.format("jdbc") \
        .option("url", "数据库连接URL") \
        .option("dbtable", table_name) \
        .option("user", "数据库账号") \
        .option("password", "数据库密码") \
        .option("numPartitions", 2) # 控制单表写入并行度，避免压垮数据库
        .mode("append") \
        .save()

# 2线程并行执行两个写表任务
with ThreadPoolExecutor(max_workers=2) as pool:
    future1 = pool.submit(write_table, attr1_df, "attribute1_table")
    future2 = pool.submit(write_table, attr2_df, "attribute2_table")
    # 等待所有任务执行完成
    future1.result()
    future2.result()

Scala环境可以用Future实现相同逻辑。

方案2：结构化流场景并行注册多个流查询 #

如果你用的是Spark Structured Streaming消费Kafka数据，直接定义多个独立的流写入查询即可，Spark会自动并行调度：

# 写入表1的流查询
query1 = attr1_df.writeStream \
    .format("jdbc") \
    .option("checkpointLocation", "/checkpoint/attr1_table") \
    .option("dbtable", "attribute1_table") \
    .trigger(processingTime="1分钟") # 按你的业务需求设置触发策略
    .start()

# 写入表2的流查询
query2 = attr2_df.writeStream \
    .format("jdbc") \
    .option("checkpointLocation", "/checkpoint/attr2_table") \
    .option("dbtable", "attribute2_table") \
    .trigger(processingTime="1分钟")
    .start()

# 阻塞等待所有流任务运行
query1.awaitTermination()
query2.awaitTermination()

注意事项 #

• 确保数据库侧的连接池配置足够支撑并行写入的连接需求，避免连接被打满导致写入失败
• 集群资源需满足两个写入作业并行运行的资源需求，否则作业会排队仍按串行执行
• 缓存的父DF使用完成后可调用parent_df.unpersist()释放缓存空间

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