如何用PySpark(Synapse)自动将复杂JSONB转为报表用行列结构？

问题背景 #

项目中PostgreSQL数据库包含上百个基于FHIR规范的jsonb类型列，存储着复杂嵌套JSON数据，需要转换为报表团队可用的扁平表格格式。PowerBI仅能完成部分自动转换，无法覆盖全部需求。当前已通过PySpark-Synapse完成以下操作：

• 通过JDBC连接获取数据
• 将jsonb列转换为JSON字符串
• 生成对应JSON Schema并展开顶层属性为列

但数组元素转成行仍需手动调用get_json_object处理，面对上百列的场景效率极低，需要全自动化的转换方案。

解决方案 #

1. 核心工具函数：递归展开数组与嵌套结构 #

编写通用函数，自动识别DataFrame中的数组字段和嵌套结构体，递归完成数组行展开与结构体列扁平化，无需手动指定字段：

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

def flatten_and_explode(df):
    while True:
        # 识别所有数组类型字段
        array_cols = [field.name for field in df.schema.fields 
                      if isinstance(field.dataType, ArrayType)]
        if not array_cols:
            break
        
        # 批量展开数组（使用explode_outer保留空值行）
        for col in array_cols:
            df = df.withColumn(col, F.explode_outer(F.col(col)))
        
        # 识别所有嵌套结构体字段
        nested_struct_cols = [field.name for field in df.schema.fields 
                              if isinstance(field.dataType, StructType)]
        if not nested_struct_cols:
            continue
        
        # 展开结构体为顶层列，添加原字段名前缀避免冲突
        for col in nested_struct_cols:
            struct_fields = df.schema[col].dataType.names
            df = df.select(
                "*", 
                *[F.col(f"{col}.{field}").alias(f"{col}_{field}") for field in struct_fields]
            ).drop(col)
    return df

2. 批量处理多FHIR json列 #

针对上百个json列的场景，编写批量处理函数，自动完成JSON字符串转结构体、顶层展开、递归扁平化的全流程：

def batch_process_fhir_json(df, json_col_names):
    for col_name in json_col_names:
        # 从JSON字符串推断Schema并转换为结构体
        sample_json = df.select(col_name).first()[0]
        json_schema = F.schema_of_json(sample_json)
        df = df.withColumn(f"{col_name}_struct", F.from_json(F.col(col_name), json_schema))
        
        # 展开结构体顶层字段（带前缀）
        struct_fields = df.schema[f"{col_name}_struct"].dataType.names
        df = df.select(
            "*", 
            *[F.col(f"{col_name}_struct.{field}").alias(f"{col_name}_{field}") for field in struct_fields]
        ).drop(col_name, f"{col_name}_struct")
    
    # 统一处理所有数组与嵌套结构
    return flatten_and_explode(df)

3. 调用示例 #

# 假设从PostgreSQL读取的原始DataFrame为raw_df
# 筛选所有FHIR json列（示例：列名以fhir_开头）
fhir_json_columns = [col for col in raw_df.columns if col.startswith("fhir_")]

# 执行批量转换
final_flat_df = batch_process_fhir_json(raw_df, fhir_json_columns)

# 输出结果
final_flat_df.show()

关键注意事项 #

• Schema兼容性：由于FHIR规范是标准化的，schema_of_json推断的Schema可适配绝大多数场景；若存在少量非标准结构，可提前手动定义统一Schema传入from_json。
• 空值保留：使用explode_outer替代explode，避免因数组为空导致数据丢失。
• 性能优化：批量处理时尽量减少DataFrame的转换次数，避免频繁Shuffle；若数据量极大，可考虑分批次处理或启用Spark的自适应执行。

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