Databricks嵌套JSON数组处理：展开/扁平化与customerId提取

术语区分 #

• JSON Explode：针对数组类型字段，将数组中的每个元素拆分为独立行，是处理数组嵌套的基础操作。
• JSON Flattening：将嵌套的结构体（如shipmentDetails、totalPrice）展开为顶级列，消除层级结构。
• JSON Unpacking：是涵盖上述两种操作的宽泛概念，指将嵌套JSON转换为扁平表格格式的整个过程。

分步处理方案 #

假设你的数据已加载到Databricks的DataFrame中，其中包含一个名为datasets的数组类型列，数组元素为订单JSON结构体。

步骤1：Explode数组列，拆分订单行 #

先将datasets数组拆分为每行一个订单：

from pyspark.sql.functions import explode

# 假设原始DataFrame名为raw_df
exploded_df = raw_df.select(explode("datasets").alias("order_data"))

步骤2：筛选指定customerId的数据 #

从拆分后的订单中提取customerId='cust5001'的记录：

from pyspark.sql.functions import col

filtered_df = exploded_df.filter(col("order_data.customerId") == "cust5001")

步骤3：扁平化所有嵌套结构 #

将订单中的嵌套结构体（shipmentDetails、totalPrice）和数组（orderDetails）全部展开：

from pyspark.sql.functions import explode, col

# 先展开orderDetails数组
flattened_step1 = filtered_df.select(
    col("order_data.customerId"),
    col("order_data.orderDate"),
    col("order_data.orderId"),
    col("order_data.shipmentDetails.*"),
    explode(col("order_data.orderDetails")).alias("order_detail")
)

# 再展开order_detail中的totalPrice结构体
final_flattened_df = flattened_step1.select(
    "*",
    col("order_detail.productId"),
    col("order_detail.quantity"),
    col("order_detail.sequence"),
    col("order_detail.totalPrice.*")
).drop("order_detail")

# 查看结果
final_flattened_df.show(truncate=False)

替代方案：使用Databricks SQL处理 #

如果习惯用SQL，可以创建临时视图后操作：

-- 创建临时视图
CREATE OR REPLACE TEMP VIEW raw_orders AS
SELECT * FROM raw_df;

-- Explode数组+筛选指定customerId
CREATE OR REPLACE TEMP VIEW exploded_filtered_orders AS
SELECT 
    explode(datasets) AS order_data
FROM raw_orders
WHERE order_data.customerId = 'cust5001';

-- 扁平化所有嵌套结构
SELECT
    order_data.customerId,
    order_data.orderDate,
    order_data.orderId,
    order_data.shipmentDetails.city,
    order_data.shipmentDetails.country,
    order_data.shipmentDetails.postalCode,
    order_data.shipmentDetails.state,
    order_data.shipmentDetails.street,
    od.productId,
    od.quantity,
    od.sequence,
    od.totalPrice.gross,
    od.totalPrice.net,
    od.totalPrice.tax
FROM exploded_filtered_orders
LATERAL VIEW explode(order_data.orderDetails) AS od;

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