使用PySpark SQL从结构体数组中按键/值查询id

针对3000万行的大型DataFrame，**避免使用explode**是提升查询效率的核心——因为explode会将每行的tags数组拆分成多行，导致数据量暴增（比如每个tags平均含5个元素，3000万行会变成1.5亿行），大幅增加计算和内存开销。下面是更高效的方案：

一、直接使用数组过滤函数（推荐） #

利用Spark内置的数组条件检查函数，不用展开数组，直接对每行的tags数组做条件判断，过滤出符合要求的行。

1. 查询包含特定key的所有id #

通过array_exists函数（Spark 3.0+支持）检查数组中是否存在匹配key的结构体：

PySpark 示例：

from pyspark.sql.functions import expr

# 替换'k1'为目标key
result = df.filter(expr("array_exists(tags, x -> x.key = 'k1')")).select("id")
result.show()

Scala 示例：

import org.apache.spark.sql.functions.expr

val result = df.filter(expr("array_exists(tags, x -> x.key = 'k1')")).select("id")
result.show()

2. 查询包含特定key-value对的所有id #

只需在条件中同时匹配key和value：

PySpark 示例：

result = df.filter(expr("array_exists(tags, x -> x.key = 'k1' AND x.value = 'v1')")).select("id")
result.show()

Scala 示例：

val result = df.filter(expr("array_exists(tags, x -> x.key = 'k1' AND x.value = 'v1')")).select("id")
result.show()

如果使用的Spark版本低于3.0，可以用array_filter替代：

# PySpark 2.x版本写法
from pyspark.sql.functions import size, array_filter

result = df.filter(size(array_filter("tags", lambda x: x.key == 'k1')) > 0).select("id")

二、关于create_map的适用性 #

不建议用create_map转换tags数组：

• 若tags数组中存在重复key，create_map会自动覆盖重复项，导致数据丢失；
• 转换map的过程需要遍历整个数组，额外增加计算开销，不如直接用数组过滤高效。

额外优化建议 #

如果这类查询是高频操作：

• 可以将DataFrame存储为Delta Lake格式，并针对tags列构建布隆索引或Z-Order索引，进一步加速过滤；
• 若tags的key集合固定，可考虑将常用key提取为单独列（比如用array_contains结合自定义逻辑），但需权衡存储开销。

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