在PySpark中利用数组列与列表交集进行DataFrame过滤的正确姿势-Python教程-PHP中文网

在PySpark中利用数组列与列表交集进行DataFrame过滤的正确姿势

花韻仙語

发布： 2025-09-23 17:22:15

原创

219人浏览过

在PySpark中利用数组列与列表交集进行DataFrame过滤的正确姿势

本文详细介绍了如何在PySpark中高效地过滤DataFrame，当需要根据数组列与一个给定Python列表的交集来筛选数据时。核心解决方案是利用pyspark.sql.functions.arrays_overlap函数，并结合lit函数将Python列表中的元素转换为Spark字面量表达式，从而构建正确的过滤条件，避免常见的AnalysisException错误。文章提供了清晰的示例代码和关键概念解释，旨在帮助用户正确实现此类复杂过滤逻辑。

1. 问题背景与挑战

在数据处理中，我们经常需要对包含数组类型列的spark dataframe进行过滤。一个常见的需求是，筛选出那些数组列中至少包含给定python列表（例如 [item1, item2, ...]）中一个或多个元素的行。

在SQL中，这种操作非常直观，通常可以使用arrays_overlap函数：

SELECT <columns>
FROM <table>
WHERE arrays_overlap(<array_column>, array(<list_elements>))

登录后复制

然而，当尝试将这种逻辑直接转换为PySpark时，许多用户会遇到困难。一个常见的错误尝试是：

from pyspark.sql.functions import col, array, arrays_overlap

# 假设 target_list 是一个 Python 列表，如 ['apple', 'banana']
df.filter(arrays_overlap(col("array_column"), array(target_list)))

登录后复制

这段代码通常会导致AnalysisException，错误信息类似于[UNRESOLVED_COLUMN.WITH_SUGGESTION] A column or function parameter with name '<1st item in list>' cannot be resolved.。这是因为array()函数在接收非列参数时，期望的是字面量表达式（literal expressions），而不是原始的Python列表元素。虽然array_contains函数可以处理单个元素，但它无法满足与整个列表进行交集判断的需求。

2. 解决方案：结合 lit 函数

解决这个问题的关键在于，将Python列表中的每个元素转换为Spark的字面量表达式（literal expression），然后再用array函数将其组合成一个字面量数组。这可以通过pyspark.sql.functions.lit函数来实现。

lit函数的作用是将一个Python值转换为一个Spark列表达式，这个表达式代表着一个常量值。当我们将列表中的每个元素都通过lit转换后，再将这些字面量表达式传递给array函数，array函数就能正确地构建一个包含这些字面量值的数组。

序列猴子开放平台

具有长序列、多模态、单模型、大数据等特点的超大规模语言模型

查看详情

正确的PySpark实现如下：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, array, arrays_overlap, lit

# 1. 初始化 SparkSession
spark = SparkSession.builder.appName("ArrayColumnFilter").getOrCreate()

# 2. 准备示例数据
data = [
    (1, ["apple", "banana", "orange"]),
    (2, ["grape", "kiwi"]),
    (3, ["banana", "strawberry"]),
    (4, ["mango", "pineapple"]),
    (5, ["apple", "grape"])
]
df = spark.createDataFrame(data, ["id", "fruits_array"])
df.printSchema()
df.show()

# 3. 定义用于过滤的 Python 列表
target_list = ["banana", "grape", "lemon"]

# 4. 构建正确的过滤条件
# 使用 map(lit, target_list) 将列表中的每个元素转换为 lit 表达式
# 使用 * 解包这些 lit 表达式作为 array 函数的参数
# 最后，使用 arrays_overlap 进行比较
filtered_df = df.filter(
    arrays_overlap(col("fruits_array"), array(*map(lit, target_list)))
)

# 5. 显示过滤结果
print(f"\n原始DataFrame：")
df.show()

print(f"\n过滤列表：{target_list}")
print("\n过滤后的DataFrame（fruits_array与target_list有交集）：")
filtered_df.show()

# 6. 停止 SparkSession
spark.stop()

登录后复制

运行结果示例：

root
 |-- id: long (nullable = true)
 |-- fruits_array: array (nullable = true)
 |    |-- element: string (nullable = true)

+---+--------------------+
| id|        fruits_array|
+---+--------------------+
|  1|[apple, banana, o...|
|  2|       [grape, kiwi]|
|  3|[banana, strawber...|
|  4|[mango, pineapple]|
|  5|       [apple, grape]|
+---+--------------------+

原始DataFrame：
+---+--------------------+
| id|        fruits_array|
+---+--------------------+
|  1|[apple, banana, o...|
|  2|       [grape, kiwi]|
|  3|[banana, strawber...|
|  4|[mango, pineapple]|
|  5|       [apple, grape]|
+---+--------------------+

过滤列表：['banana', 'grape', 'lemon']

过滤后的DataFrame（fruits_array与target_list有交集）：
+---+--------------------+
| id|        fruits_array|
+---+--------------------+
|  1|[apple, banana, o...|
|  2|       [grape, kiwi]|
|  3|[banana, strawber...|
|  5|       [apple, grape]|
+---+--------------------+

登录后复制

从结果可以看出，id为1、2、3、5的行被保留，因为它们的fruits_array列与["banana", "grape", "lemon"]存在交集（例如，id=1包含"banana"，id=2包含"grape"，id=3包含"banana"，id=5包含"grape"）。

3. 关键概念与注意事项

arrays_overlap(array1, array2): 这个函数用于判断两个数组是否有共同的元素。如果存在任何共同元素，则返回True；否则返回False。它是进行数组交集判断的核心。
lit(value): lit函数将一个Python字面量（如字符串、数字、布尔值）转换为一个Spark SQL的字面量列。这是在PySpark中构建常量值表达式的常用方法。
*`array(expressions)**:array`函数有两种主要用法：
- 当参数是列名时，它将这些列的值组合成一个新的数组列。例如：array(col("col1"), col("col2"))。
- 当参数是字面量表达式时，它会创建一个包含这些字面量值的字面量数组。例如：array(lit("a"), lit("b"))。
- 我们这里的解决方案属于第二种情况，map(lit, target_list)生成了一系列字面量表达式，*操作符将它们解包作为array函数的独立参数。
错误避免: 理解array函数对参数类型的期望是避免AnalysisException的关键。直接传递Python列表array(target_list)会被Spark误解为target_list中的第一个元素是一个列名，因此无法解析。

4. 总结

在PySpark中，当需要使用一个Python列表与DataFrame的数组列进行交集过滤时，务必记住使用pyspark.sql.functions.lit函数将列表中的每个元素转换为Spark字面量表达式。然后，通过array(*map(lit, your_list))的方式构建一个字面量数组，并将其作为arrays_overlap函数的第二个参数。这种模式是处理这类复杂数组过滤逻辑的标准且正确的方法，能够确保代码的健壮性和准确性。

以上就是在PySpark中利用数组列与列表交集进行DataFrame过滤的正确姿势的详细内容，更多请关注php中文网其它相关文章！