如何将Delta Live Table中的JSON字符串字段扩展为新DLT？

Python实现方案 #

先从源DLT表抽取JSON样本自动推断Schema，再在目标DLT表中用该Schema解析并展开字段：

import dlt
from pyspark.sql.functions import from_json, col

# 1. 预推断JSON Schema（仅需初始化时执行一次，若Schema稳定可固化）
source_df = spark.read.table("your_source_dlt_table")
json_schema = spark.read.json(source_df.select("JsonData").rdd.map(lambda row: row[0])).schema

# 2. 定义目标DLT表
@dlt.table(
  name="parsed_json_dlt_table",
  comment="解析JSON字段后的DLT表"
)
def create_parsed_table():
    # 根据源表类型选择read或read_stream
    source_data = dlt.read_stream("your_source_dlt_table") if source_is_streaming else dlt.read("your_source_dlt_table")
    # 过滤无效JSON行
    filtered_data = source_data.filter(col("JsonData").isNotNull())
    # 解析JSON并展开所有字段
    parsed_data = filtered_data.withColumn("parsed_json", from_json(col("JsonData"), json_schema)) \
                               .select("*", "parsed_json.*") \
                               .drop("parsed_json")
    return parsed_data

说明：若源JSON结构会动态变更，需定期重新推断Schema；DLT更适配固定Schema场景，结构频繁变动建议增加Schema校验逻辑。

SQL实现方案 #

通过Spark SQL生成JSON Schema，再在DLT表定义中解析并展开：

1. 预生成JSON Schema：

CREATE OR REPLACE TEMP VIEW json_schema_view AS
SELECT schema_of_json(collect_list(JsonData)[0]) AS json_schema
FROM your_source_dlt_table
WHERE JsonData IS NOT NULL
LIMIT 1;

2. 创建流式/批处理DLT表：

-- 流式表示例
CREATE OR REFRESH STREAMING TABLE parsed_json_dlt_table
COMMENT '解析JSON字段后的DLT表'
AS
SELECT 
  *,
  inline(array(from_json(JsonData, (SELECT json_schema FROM json_schema_view))))
FROM your_source_dlt_table
WHERE JsonData IS NOT NULL;

说明：schema_of_json依赖单个JSON字符串生成Schema，若JSON结构多样，建议用Python方式合并多样本Schema。

常见问题说明 #

• from_json失败原因：必须明确指定Schema，未提供时无法自动推断，需先通过样本获取Schema；
• parse_json局限：仅将字符串转为STRUCT类型，需手动指定要展开的字段，无法自动批量展开；
• json.loads不适用：属于Python原生函数，无法直接在Spark分布式环境中使用，用UDF实现会导致性能下降且无法自动推断Schema。

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