如何对比两个PySpark DataFrame的Schema并生成对比表

需求说明 #

拥有两个PySpark DataFrame，需生成包含列名、df1数据类型、df2数据类型的对比DataFrame，并新增列判断两者数据类型是否一致。已知两个Schema不相等，解析printSchema()输出为表格存在困难。

两个DataFrame的printSchema()结果如下：
df1的Schema：

root
 |-- orgID: string (nullable = true)
 |-- deptID: string (nullable = true)
 |-- systemID: string (nullable = true)
 |-- eventId: string (nullable = true)
 |-- eventType: string (nullable = true)
 |-- autoID: string (nullable = true)
 |-- personID: string (nullable = true)
 |-- employeeFirst: string (nullable = true)
 |-- employeeMiddle: string (nullable = true)
 |-- employeeLast: string (nullable = true)
 |-- employeeDOB: string (nullable = true)

df2的Schema：

root
 |-- orgID: integer (nullable = true)
 |-- deptID: string (nullable = true)
 |-- systemID: string (nullable = true)
 |-- eventId: integer (nullable = true)
 |-- eventType: string (nullable = true)
 |-- autoID: string (nullable = true)
 |-- personID: integer (nullable = true)
 |-- employeeFirst: string (nullable = true)
 |-- employeeMiddle: string (nullable = true)
 |-- employeeLast: string (nullable = true)
 |-- employeeDOB: timestamp (nullable = false)

期望生成的对比表格式：

+-------------+----------+---------+-----------+
|       column|   df1type|  df2type|types_match|
+-------------+----------+---------+-----------+
|        orgID|    string|  integer|      false|
|       deptID|    string|   string|       true|
...
|  employeeDOB|    string|timestamp|      false|
+-------------+----------+---------+-----------+

实现方案 #

无需解析printSchema()的输出，直接操作DataFrame的schema属性（本质是StructType对象，包含多个StructField元素，每个元素可直接获取列名、数据类型等信息），步骤如下：

代码实现 #

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, BooleanType

# 假设已初始化SparkSession，df1和df2为目标DataFrame
spark = SparkSession.builder.appName("SchemaComparison").getOrCreate()

# 1. 将两个Schema转换为字典：键为列名，值为数据类型字符串
df1_schema_dict = {field.name: str(field.dataType) for field in df1.schema.fields}
df2_schema_dict = {field.name: str(field.dataType) for field in df2.schema.fields}

# 2. 获取所有列名的并集（避免遗漏仅在单个DataFrame中存在的列）
all_columns = sorted(set(df1_schema_dict.keys()).union(set(df2_schema_dict.keys())))

# 3. 构建对比数据列表
comparison_data = []
for col in all_columns:
    df1_type = df1_schema_dict.get(col, None)
    df2_type = df2_schema_dict.get(col, None)
    types_match = df1_type == df2_type
    comparison_data.append((col, df1_type, df2_type, types_match))

# 4. 定义对比结果DataFrame的Schema
comparison_schema = StructType([
    StructField("column", StringType(), nullable=False),
    StructField("df1type", StringType(), nullable=True),
    StructField("df2type", StringType(), nullable=True),
    StructField("types_match", BooleanType(), nullable=False)
])

# 5. 创建对比结果DataFrame
comparison_df = spark.createDataFrame(comparison_data, schema=comparison_schema)

# 查看结果
comparison_df.show()

说明 #

• 直接操作schema属性比解析printSchema()输出更可靠，避免因输出格式变化导致的解析错误
• 处理了仅在单个DataFrame中存在的列，这类列的另一方类型会显示为null，types_match为false
• 最终生成的comparison_df包含所有需求列，可直接用于查看或进一步处理

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