代码之家 › 专栏 › 技术社区 › ZygD

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

conditional-statements apache-spark-sql pyspark apache-spark python

0

ZygD · 技术社区 · 4 年前

数据

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”,“|”表示“or”,“~”表示“not”。

0 回复 | 直到 4 年前

1

AdibP 4 年前

您可以使用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]

2

0

ZygD 4 年前

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

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

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