PySpark Databricks中如何将统计结果DataFrame转为整数并构建变更报告？

一、将统计结果DataFrame转为整数 #

你通过sum()得到的是单一行单列的DataFrame，要提取整数只需将结果从分布式DataFrame拉取到Driver节点即可，常用方法如下：

方法1：使用first()（推荐） #

# 假设你的统计DF是Change_To_Value1，聚合后的列名为sum(change_flag1)
change_val1 = Change_To_Value1.first()[0] if Change_To_Value1.count() > 0 else 0

first()直接取第一行数据，效率比collect()更高，适合单个统计值的提取。

方法2：处理空值更严谨的写法 #

如果统计结果可能为null（比如没有符合条件的变更），可以用coalesce确保返回整数：

from pyspark.sql import functions as F

# 先将统计列转为int并替换null为0，再提取
change_val1 = Change_To_Value1.select(F.coalesce(F.col("sum(change_flag1)").cast("int"), F.lit(0))).first()[0]

二、更高效的变更报告构建方法 #

单独统计每个列再拼接报告的方式不够高效，推荐直接通过全外连接+行/列级标记+聚合一次性生成完整报告，适配大数据场景且无需手动转换整数：

步骤1：关联新旧DataFrame并标记行级变更 #

假设新旧DataFrame为old_df（数据库已加载）、new_df（新文件），主键为id：

from pyspark.sql import functions as F

# 全外连接关联主键
joined_df = old_df.join(new_df, on="id", how="full_outer")

# 标记行状态：新增/删除/修改
row_change_df = joined_df.withColumn(
    "row_status",
    F.when(F.col("old_df.id").isNull(), "新增")
    .when(F.col("new_df.id").isNull(), "删除")
    .otherwise("修改")
)

步骤2：生成列级变更标记并聚合统计 #

# 排除主键，获取需要对比的列列表
compare_cols = [col for col in old_df.columns if col != "id"]

# 为每个对比列生成变更标记（1=变更，0=未变更）
for col in compare_cols:
    row_change_df = row_change_df.withColumn(
        f"{col}_changed",
        F.when(F.col(f"old_df.{col}") != F.col(f"new_df.{col}"), 1).otherwise(0)
    )

# 聚合生成完整变更报告
report_df = row_change_df.groupBy().agg(
    # 行级变更统计
    F.count(F.when(F.col("row_status") == "新增", True)).alias("新增行数"),
    F.count(F.when(F.col("row_status") == "删除", True)).alias("删除行数"),
    F.count(F.when(F.col("row_status") == "修改", True)).alias("修改行数"),
    # 列级变更统计
    *[F.sum(f"{col}_changed").alias(f"{col}_变更次数") for col in compare_cols]
)

# 查看结果
report_df.display()  # Databricks环境直接用display展示

这种方法直接生成结构化的变更报告DataFrame，无需手动处理单个统计值，结果可直接与SSIS包输出对比。

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