PySpark中如何将已有DataFrame的Schema应用到缺失列的DataFrame

PySpark 解决方案 #

方法1：手动构造嵌套结构体 #

直接对照目标Schema（df_1的Schema），把df_2的字段映射到对应嵌套层级，缺失字段统一填充null：

from pyspark.sql import functions as F

# 构造association_info结构体，仅保留df_2的study_id，其余字段设为null
association_info = F.struct(
    F.lit(None).cast("array<string>").alias("ancestry"),
    F.lit(None).cast("string").alias("doi"),
    F.lit(None).cast("string").alias("gwas_catalog_id"),
    F.lit(None).cast("double").alias("neg_log_pval"),
    F.col("study_id").alias("study_id"),
    F.lit(None).cast("string").alias("pubmed_id"),
    F.lit(None).cast("string").alias("url")
)

# 构造evidence数组内的结构体，仅保留df_2的description，其余字段设为null
evidence_element = F.struct(
    F.lit(None).cast("string").alias("class"),
    F.lit(None).cast("string").alias("confidence"),
    F.lit(None).cast("string").alias("curated_by"),
    F.col("description").alias("description"),
    F.lit(None).cast("string").alias("pubmed_id"),
    F.lit(None).cast("string").alias("source")
)

# 构造gold_standard_info结构体，保留df_2的gene_id，其余字段设为null
gold_standard_info = F.struct(
    F.array(evidence_element).alias("evidence"),
    F.col("gene_id").alias("gene_id"),
    F.lit(None).cast("string").alias("highest_confidence")
)

# 转换df_2并保留目标嵌套结构
df_2_transformed = df_2.select(
    association_info.alias("association_info"),
    gold_standard_info.alias("gold_standard_info")
)

# 验证Schema是否匹配
df_2_transformed.printSchema()

方法2：递归生成通用转换逻辑 #

如果Schema层级多、字段繁杂，手动构造效率低，可以写递归函数自动遍历目标Schema，生成对应转换表达式：

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, ArrayType

def generate_schema_expr(schema, parent_path=""):
    exprs = []
    for field in schema.fields:
        field_path = f"{parent_path}.{field.name}" if parent_path else field.name
        if isinstance(field.dataType, StructType):
            # 递归处理结构体字段
            struct_exprs = generate_schema_expr(field.dataType, field_path)
            exprs.append(F.struct(*struct_exprs).alias(field.name))
        elif isinstance(field.dataType, ArrayType):
            if isinstance(field.dataType.elementType, StructType):
                # 特殊处理gold_standard_info.evidence的字段映射
                if field_path == "gold_standard_info.evidence":
                    evidence_element = F.struct(
                        F.lit(None).cast("string").alias("class"),
                        F.lit(None).cast("string").alias("confidence"),
                        F.lit(None).cast("string").alias("curated_by"),
                        F.col("description").alias("description"),
                        F.lit(None).cast("string").alias("pubmed_id"),
                        F.lit(None).cast("string").alias("source")
                    )
                    exprs.append(F.array(evidence_element).alias(field.name))
                else:
                    # 其他数组嵌套结构填充null
                    exprs.append(F.lit(None).cast(field.dataType).alias(field.name))
            else:
                # 基础数组类型填充null
                exprs.append(F.lit(None).cast(field.dataType).alias(field.name))
        else:
            # 基础数据类型：优先用df_2已有字段，否则填充null
            if field.name in df_2.columns:
                exprs.append(F.col(field.name).alias(field.name))
            else:
                exprs.append(F.lit(None).cast(field.dataType).alias(field.name))
    return exprs

# 基于df_1的Schema生成转换表达式
target_schema = df_1.schema
transform_exprs = generate_schema_expr(target_schema)

# 转换df_2
df_2_transformed = df_2.select(*transform_exprs)

Pandas 解决方案 #

通过构造嵌套字典的方式，将df_2的字段映射到目标嵌套层级：

import pandas as pd

def build_nested_row(row):
    return {
        "association_info": {
            "ancestry": None,
            "doi": None,
            "gwas_catalog_id": None,
            "neg_log_pval": None,
            "study_id": row["study_id"],
            "pubmed_id": None,
            "url": None
        },
        "gold_standard_info": {
            "evidence": [
                {
                    "class": None,
                    "confidence": None,
                    "curated_by": None,
                    "description": row["description"],
                    "pubmed_id": None,
                    "source": None
                }
            ],
            "gene_id": row["gene_id"],
            "highest_confidence": None
        }
    }

# 转换df_2为嵌套结构
df_2_transformed = pd.DataFrame([build_nested_row(row) for _, row in df_2.iterrows()])

# 查看结构详情
print(df_2_transformed.dtypes)
print(df_2_transformed.iloc[0])

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