使用PySpark提取嵌套Map结构中的指定字段

原始数据Schema #

sitename: string (nullable = true)
 |-- publisherUid: string (nullable = true)
 |-- requestid: string (nullable = true)
 |-- deviceType: string (nullable = true)
 |-- vscore: map (nullable = true)
 |    |-- key: string
 |    |-- value: map (valueContainsNull = true)
 |    |    |-- key: string
 |    |    |-- value: string (valueContainsNull = true)

需求 #

提取以下字段：

• sitename
• publisherUid
• requestid
• deviceType
• vscore的外层键（命名为vscore_key）
• 内层Map的第一个值内容
• 内层Map的第二个值内容

解决方案 #

方式1：Spark DataFrame API #

通过explode_outer拆分外层Map，再将内层Map转为数组提取指定位置的值：

import org.apache.spark.sql.functions._

// 假设原始DataFrame名为df
val resultDF = df
  // 拆分外层vscore Map为键值对，保留vscore为null的行
  .select(
    col("sitename"),
    col("publisherUid"),
    col("requestid"),
    col("deviceType"),
    explode_outer(col("vscore")).alias("vscore_key", "inner_map")
  )
  // 将内层Map转为值数组
  .withColumn("inner_map_values", map_values(col("inner_map")))
  // 提取数组中第一个和第二个值
  .withColumn("first_inner_value", col("inner_map_values").getItem(0))
  .withColumn("second_inner_value", col("inner_map_values").getItem(1))
  // 筛选最终需要的字段
  .select(
    "sitename", "publisherUid", "requestid", "deviceType",
    "vscore_key", "first_inner_value", "second_inner_value"
  )

方式2：Spark SQL #

先创建临时视图，再通过SQL语句完成提取：

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

-- 提取目标字段
SELECT
  sitename,
  publisherUid,
  requestid,
  deviceType,
  vscore_key,
  inner_map_values[0] AS first_inner_value,
  inner_map_values[1] AS second_inner_value
FROM (
  SELECT
    sitename,
    publisherUid,
    requestid,
    deviceType,
    -- 拆分外层Map，保留vscore为null的行
    explode_outer(vscore) AS (vscore_key, inner_map),
    -- 将内层Map转为值数组
    map_values(inner_map) AS inner_map_values
  FROM temp_data
);

注意事项 #

• 如果内层Map的键是固定值，可直接通过键名取值，比如inner_map['固定键名']，无需转为数组。
• 使用explode_outer替代explode，可以避免丢失vscore为null的原始数据行。

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