PySpark：过滤Struct列内嵌套数组指定元素并保留上层结构

核心思路：不直接提取数组过滤，而是重新构建整个上层Struct结构，对Struct内的每个Map做值转换，只过滤Map中数组的目标元素。

适用于Spark 3.0+版本（推荐） #

Spark 3.0新增的transform_values可以直接修改Map的Value部分，无需拆解重组Map，代码更简洁：

from pyspark.sql import functions as F

# 定义要过滤的storeId（注意是string类型，需加引号）
target_store_id = "123124124"

# 处理first_profiles：遍历Map的每个数组Value，过滤掉目标storeId的元素
processed_first = F.transform_values(
    F.col("my_feature_name.first_profiles"),
    lambda arr: F.filter(arr, lambda x: x.storeId != target_store_id)
)

# 同理处理second_profile
processed_second = F.transform_values(
    F.col("my_feature_name.second_profile"),
    lambda arr: F.filter(arr, lambda x: x.storeId != target_store_id)
)

# 重新构建完整的my_feature_name Struct，替换原列
df = df.withColumn(
    "my_feature_name",
    F.struct(processed_first.alias("first_profiles"), processed_second.alias("second_profile"))
)

适用于Spark 2.x版本 #

如果无法升级Spark版本，可以通过map_entries拆解Map、处理后再重组：

from pyspark.sql import functions as F

target_store_id = "123124124"

# 处理first_profiles：拆解Map→过滤数组→重组Map
processed_first = F.map_from_entries(
    F.transform(
        F.map_entries(F.col("my_feature_name.first_profiles")),
        lambda entry: F.struct(
            entry.key.alias("key"),
            F.filter(entry.value, lambda x: x.storeId != target_store_id).alias("value")
        )
    )
)

# 处理second_profile同理
processed_second = F.map_from_entries(
    F.transform(
        F.map_entries(F.col("my_feature_name.second_profile")),
        lambda entry: F.struct(
            entry.key.alias("key"),
            F.filter(entry.value, lambda x: x.storeId != target_store_id).alias("value")
        )
    )
)

# 重新构建Struct
df = df.withColumn(
    "my_feature_name",
    F.struct(processed_first.alias("first_profiles"), processed_second.alias("second_profile"))
)

过滤多个storeId的情况 #

如果需要排除多个storeId，只需修改过滤条件：

# 示例：排除storeId为"123124124"和"567890"的元素
lambda x: x.storeId NOT IN ("123124124", "567890")

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