关于Databricks Delta Live Tables中嵌套JSON解析与加载的技术疑问

一、通过SQL Notebook解析嵌套JSON并加载到Delta Live Table #

直接在Delta Live Table（DLT）的SQL定义中利用Spark SQL的JSON解析函数即可处理嵌套结构，具体实现如下：

步骤示例： #

1. 创建原始JSON源表（支持批处理或流式读取，显式指定Schema提升性能）
2. 解析嵌套字段并生成DLT表

-- 1. 定义嵌套JSON的Schema并读取原始数据
CREATE LIVE TABLE raw_nested_json
COMMENT "存储原始嵌套JSON的源表"
AS SELECT * FROM cloud_files(
  "/path/to/your/json/directory",
  "json",
  map("schema", "struct<id:int, profile:struct<username:string, phone:string>, transactions:array<struct<txn_id:string, amount:double, timestamp:string>>>")
);

-- 2. 解析嵌套JSON并创建DLT表
CREATE LIVE TABLE parsed_dlt_sql
COMMENT "SQL解析后的Delta Live Table"
AS SELECT
  id,
  profile.username AS user_name,
  profile.phone AS user_phone,
  -- 展开数组类型的嵌套字段，若不需要展开可直接保留数组
  explode(transactions) AS transaction_details
FROM raw_nested_json;

如果JSON内容存储在字符串列中，使用from_json函数解析：

CREATE LIVE TABLE parsed_string_json_dlt
AS SELECT
  id,
  from_json(profile_json, "struct<username:string, phone:string>").username AS user_name,
  from_json(profile_json, "struct<username:string, phone:string>").phone AS user_phone
FROM raw_json_string_table;

二、将Python Notebook解析后的数据加载到Delta Live Table #

若已在Python Notebook中完成嵌套JSON解析并得到Spark DataFrame，可通过以下两种方式写入DLT：

方法1：注册临时视图后通过SQL创建DLT表 #

在Python Notebook中注册解析后的DataFrame为临时视图：

# 假设parsed_df是你解析完成的Spark DataFrame
parsed_df.createOrReplaceTempView("parsed_json_temp_view")

随后在DLT的SQL Notebook中引用该视图创建表：

CREATE LIVE TABLE parsed_dlt_python
COMMENT "Python解析后的Delta Live Table"
AS SELECT * FROM parsed_json_temp_view;

方法2：直接使用Python DLT API定义表 #

如果你的DLT Pipeline支持Python脚本，可直接在DLT代码中集成解析逻辑并生成表：

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

@dlt.table(comment="Python解析嵌套JSON生成的DLT表")
def python_parsed_dlt_table():
    # 读取原始JSON数据（若已在Notebook解析，可替换为读取临时存储的Delta文件）
    raw_df = spark.read.json("/path/to/raw/json/files")
    # 复用你Notebook中的解析逻辑
    parsed_df = raw_df.select(
        col("id"),
        col("profile.username").alias("user_name"),
        col("profile.phone").alias("user_phone"),
        explode(col("transactions")).alias("transaction_details")
    )
    return parsed_df

补充：独立Notebook解析后的数据写入DLT #

若解析工作在独立Python Notebook中完成，可先将DataFrame写入临时Delta路径，再在DLT中读取：

# Python Notebook中：将解析后的数据写入临时Delta目录
parsed_df.write.format("delta").mode("overwrite").save("/tmp/parsed_json_temp")

DLT中读取并创建表：

CREATE LIVE TABLE external_parsed_dlt
AS SELECT * FROM delta.`/tmp/parsed_json_temp`;

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