Azure Synapse PySpark：从Schema文件加载预设Schema实现多数据集参数化

核心思路 #

把每个数据集的Schema以可读取格式存储在数据湖，通过参数化指定Schema文件路径、落地数据路径，动态加载Schema后读取JSON数据，保证数据类型准确，最终转换为Parquet格式写入Staging层。

一、Schema文件的存储格式选择 #

1. JSON格式Schema（推荐） #

将PySpark Schema序列化为JSON字符串存储，示例Schema文件内容（如user_schema.json）：

{
  "type": "struct",
  "fields": [
    {"name": "user_id", "type": "integer", "nullable": false},
    {"name": "username", "type": "string", "nullable": true},
    {"name": "register_time", "type": "timestamp", "nullable": true},
    {"name": "is_active", "type": "boolean", "nullable": false}
  ]
}

2. DDL字符串格式 #

将Schema以DDL语句存储为文本文件（如order_schema.txt）：

order_id INT NOT NULL, order_no STRING, amount DECIMAL(10,2), create_time TIMESTAMP

二、加载Schema文件为PySpark Schema对象 #

1. 加载JSON格式Schema #

from pyspark.sql.types import StructType
import json

def load_schema_from_json(schema_file_path):
    # 从数据湖读取JSON Schema文件
    schema_json = spark.read.text(schema_file_path).first()[0]
    # 转换为StructType对象
    return StructType.fromJson(json.loads(schema_json))

2. 加载DDL格式Schema #

def load_schema_from_ddl(schema_file_path):
    # 从数据湖读取DDL文本文件
    ddl_str = spark.read.text(schema_file_path).first()[0]
    # 转换为StructType对象
    return spark.sql(f"SELECT * FROM VALUES ({ddl_str})").schema

三、参数化处理多数据集的完整流程 #

# 定义参数（可通过Notebook参数控件动态传入）
varLanding = "/data-lake/landing/user_data"  # 落地层JSON数据路径
varSchema = "/data-lake/schemas/user_schema.json"  # 对应Schema文件路径
varStaging = "/data-lake/staging/user_data_parquet"  # Staging层Parquet输出路径

# 加载Schema（根据实际存储格式选择对应方法）
dataSchema = load_schema_from_json(varSchema)
# 若用DDL格式则替换为：dataSchema = load_schema_from_ddl(varSchema)

# 读取落地层JSON数据（强制使用预设Schema保证类型准确）
df = spark.read.load(varLanding, format='json', schema=dataSchema)
display(df.limit(5))

# 写入Staging层Parquet（可按需添加分区、压缩等配置）
df.write.mode("overwrite").parquet(varStaging)

注意事项 #

• 确保Schema文件与落地数据集结构严格匹配，避免因字段类型不匹配导致加载失败
• 增量数据场景可在参数中添加时间分区参数，实现增量加载与写入
• 可将参数封装为Notebook输入控件（如Databricks Widgets），更便捷切换不同数据集处理

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