使用 AND 有效地组合列表中的逻辑条件

Question

数据：

from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.getOrCreate()

data = [(1, 8, 1, None),
        (1, 1, 8, None),
        (2, 1, 8, None),
        (2, 8, 8, 9999)]
df = spark.createDataFrame(data, ['c1', 'c2', 'c3', 'c4'])

条件列表：

conditions = [
    ((F.col('c1') == 2) | (F.col('c2') < 5)),
    ((F.col('c3') == 1) | F.col('c4').isNotNull())
]

要重写的过滤器：

df = df.filter(conditions[0] & conditions[1])

df.show()
#  +---+---+---+----+
#  | c1| c2| c3|  c4|
#  +---+---+---+----+
#  |  2|  8|  8|9999|
#  +---+---+---+----+

在我的原始数据中，列表中有未定义数量的条件。我正在寻找一种使用 AND 将它们组合在一起的方法，这样我就不需要使用索引号单独引用它们了。

我试过Python的内置all()没有成功：
df = df.filter(all(conditions))

ValueError：无法将列转换为布尔值：请使用 '&' 表示 'and'、'|'在构建 DataFrame 布尔表达式时，for 'or', '~' for 'not'。

Answer 1

一种方法是将条件转换为纯 Spark，然后使用 ' AND '.join()：

conditions = [
    '((c1 = 2) OR (c2 < 5))',
    '((c3 = 1) OR isnotnull(c4))'
]

df = df.filter(' AND '.join(conditions))

Answer 2

您可以使用 Python 的 reduce 函数。

from functools import  reduce

df_result = reduce(lambda dfx, i: dfx.filter(conditions[i]), range(len(conditions)), df)

虽然它看起来像一个递归操作，但 Spark 将通过将这些过滤器与 AND 运算符组合来优化执行。

df_result.explain(False) # set to True for details

# == Physical Plan ==
# *(1) Filter (((c1#0L = 2) OR (c2#1L < 5)) AND ((c3#2L = 1) OR isnotnull(c4#3L)))
# +- *(1) Scan ExistingRDD[c1#0L,c2#1L,c3#2L,c4#3L]