PySpark如何从预定义DataSchema获取字段数以检测schema变更？

1. 获取预定义Schema的字段数量 #

你已从ADLS的JSON文件成功加载dataSchema（StructType类型），直接通过len()方法即可获取字段总数：

# 获取预定义schema的字段数量
predefined_field_count = len(dataSchema.fields)

2. 加载落地数据并获取列数 #

假设落地数据为JSON格式（可根据实际文件类型调整读取逻辑），加载为DataFrame后，通过len(df.columns)获取列数：

# 加载落地容器中的数据（路径按需调整）
landing_data_path = 'abfss://landing@[hidden].dfs.core.windows.net/'+parSourceSystemName+'/target_data/'
df_landing = spark.read.json(landing_data_path)

# 获取落地数据的列数
landing_column_count = len(df_landing.columns)

3. 对比数量并判断Schema变更 #

直接对比两个数值，输出变更判断结果：

# 字段数量对比逻辑
if predefined_field_count == landing_column_count:
    print("未检测到字段数量变更")
elif predefined_field_count < landing_column_count:
    print(f"源系统新增字段：落地数据比预定义schema多{landing_column_count - predefined_field_count}个字段")
else:
    print(f"源系统减少字段：落地数据比预定义schema少{predefined_field_count - landing_column_count}个字段")

补充：更严谨的Schema校验（可选） #

仅对比字段数量可能遗漏字段名称变更、数据类型变更的情况，若需全面校验，可遍历字段逐一比对：

# 获取预定义schema的字段名集合
predefined_fields = set(field.name for field in dataSchema.fields)
# 获取落地数据的列名集合
landing_columns = set(df_landing.columns)

# 检测新增字段
added_fields = landing_columns - predefined_fields
if added_fields:
    print(f"新增字段：{', '.join(added_fields)}")

# 检测缺失字段
missing_fields = predefined_fields - landing_columns
if missing_fields:
    print(f"缺失字段：{', '.join(missing_fields)}")

# 检测数据类型变更
for field in dataSchema.fields:
    if field.name in landing_columns:
        landing_dtype = df_landing.schema[field.name].dataType
        if landing_dtype != field.dataType:
            print(f"字段{field.name}数据类型变更：预定义为{field.dataType}，落地数据为{landing_dtype}")

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