使用PySpark遍历数据库查询200+表并合并结果至单个CSV

哥们，我来给你捋捋这个问题怎么解决最靠谱。你现在碰到的两个核心痛点：循环写入CSV只剩最后一条记录（大概率是每次重命名part文件时覆盖了之前的）、迭代处理DataFrame效率低，其实可以用一个更贴合Spark分布式特性的方案搞定——把所有表的查询结果合并成带表名的统一DataFrame，一次性写入CSV，既解决覆盖问题，又能大幅提升效率。

具体步骤&代码示例 #

1. 定义单表处理函数 #

先写个小函数，输入表名就返回包含表名和去重后errors值的DataFrame——关键是要给每条结果加上表名字段，这样最终CSV里能明确关联到对应的表。

def process_single_table(table_name):
    # 查询时新增table_name字段，同时保留distinct的errors值
    df = spark.sql(f"""
        SELECT 
            '{table_name}' AS table_name, 
            DISTINCT errors AS error_value 
        FROM {database}.{table_name} 
        WHERE errors IS NOT NULL
    """)
    return df

2. 批量生成并合并所有表的DataFrame #

用列表推导式生成所有表的DataFrame列表，再通过union合并成一个大的DataFrame。Spark 2.0+之后union和unionAll功能一致，推荐用union。

list_tables = ['accounting', 'sales', ...]  # 你的表列表

# 生成所有表的DataFrame列表
table_dfs = [process_single_table(table) for table in list_tables]

# 合并所有DataFrame（基础写法，兼容所有Spark版本）
combined_df = table_dfs[0]
for df in table_dfs[1:]:
    combined_df = combined_df.union(df)

# 如果你用Spark 3.0+，可以用更简洁的写法：
# from functools import reduce
# from pyspark.sql import DataFrame
# combined_df = reduce(DataFrame.union, table_dfs)

3. 一次性写入CSV #

合并完成后只需要写入一次CSV，再重命名part文件即可——这样就不会出现循环写入时的覆盖问题，而且Spark能优化整个执行流程，效率比循环处理高很多。

# 写入CSV（这里用overwrite模式，因为只写一次）
combined_df.repartition(1).write.mode("overwrite").option("header", "true").csv("s3://your-temp-output-path/")

# 只需要执行一次重命名操作
rename_part_file("s3://your-temp-output-path/", "final_errors_stats.csv", "s3://your-final-path/")

为什么这个方案更优？ #

• 解决覆盖问题：一次性写入只会生成一个part文件（因为repartition(1)），重命名一次就搞定，不会像循环那样每次重命名都覆盖之前的文件。
• 贴合Spark优化逻辑：迭代处理单个DataFrame会多次触发Spark作业，开销极大；合并成一个DataFrame后，Spark可以生成更高效的执行计划，减少作业提交次数，200多张表的处理速度会快很多。
• 结果更规范：最终CSV里每条记录都带有表名和对应的error值，完全符合你要的统计结果格式。

注意事项 #

• 200多张表直接合并完全没问题，因为每个表只取distinct的errors，数据量不会太大；如果担心内存，可以分批次合并，但一般没必要。
• 确保database变量已经正确定义，避免SQL语法错误。
• 如果某些表没有符合条件的errors（查询结果为空），union操作会自动忽略空DataFrame，不影响最终结果。

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