使用类似 SQL 的 IN 子句过滤 Pyspark DataFrame

Posted 2023-02-23

【中文标题】使用类似 SQL 的 IN 子句过滤 Pyspark DataFrame

【英文标题】：Filtering a Pyspark DataFrame with SQL-like IN clause

【发布时间】：2016-06-22 14:21:24

【问题描述】：

我想用类似 SQL 的IN 子句过滤 Pyspark DataFrame，如

sc = SparkContext()
sqlc = SQLContext(sc)
df = sqlc.sql('SELECT * from my_df WHERE field1 IN a')

其中a 是元组(1, 2, 3)。我收到此错误：

java.lang.RuntimeException: [1.67] failure: ``('' 预期但发现标识符 a

这基本上是说它期待像 '(1, 2, 3)' 而不是 a. 问题是我无法在 a 中手动写入值，因为它是从另一个作业中提取的。

在这种情况下我将如何过滤？

【参考方案1】：

只是一点点补充/更新：

choice_list = ["foo", "bar", "jack", "joan"]

如果你想过滤你的数据框“df”，这样你想保留基于列“v”的行，只取choice_list中的值，那么

from pyspark.sql.functions import col

df_filtered = df.where( ( col("v").isin (choice_list) ) )

【参考方案2】：

您传递给SQLContext 的字符串在SQL 环境的范围内进行评估。它没有捕获关闭。如果你想传递一个变量，你必须明确地使用字符串格式：

df = sc.parallelize([(1, "foo"), (2, "x"), (3, "bar")]).toDF(("k", "v"))
df.registerTempTable("df")
sqlContext.sql("SELECT * FROM df WHERE v IN 0".format(("foo", "bar"))).count()
##  2 

显然，出于安全考虑，这不是您在“真实”SQL 环境中使用的东西，但在这里应该无关紧要。

在实践中DataFrame DSL 是您想要创建动态查询时更好的选择：

from pyspark.sql.functions import col

df.where(col("v").isin("foo", "bar")).count()
## 2

它很容易为您构建、编写和处理 HiveQL / Spark SQL 的所有细节。

【讨论】：

对于第二种方法，你可以通过 df.where(df.v.isin("foo", "bar")).count() 来达到同样的效果

可以，但我个人不喜欢这种方法。使用col，我可以轻松地解耦SQL 表达式和特定的DataFrame 对象。因此，例如，您可以保留有用的表达字典，并在需要时选择它们。使用显式 DF 对象，您必须将其放在函数中，并且它的组合并不好。

如何用一个元组列表来做到这一点？如果我有例如[(1,1), (1,2), (1,3)] 例如，其中一个是 aid，另一个是 bid。它必须类似于col(['aid', 'bid]).isin([(1,1), (1,2)])

【参考方案3】：

您也可以对整数列执行此操作：

df_filtered = df.filter("field1 in (1,2,3)")

或者这个对于字符串列：

df_filtered = df.filter("field1 in ('a','b','c')")

【参考方案4】：

对我有用的一种稍微不同的方法是使用自定义过滤器功能进行过滤。

def filter_func(a):
"""wrapper function to pass a in udf"""
    def filter_func_(col):
    """filtering function"""
        if col in a.value:
            return True

    return False

return udf(filter_func_, BooleanType())

# Broadcasting allows to pass large variables efficiently
a = sc.broadcast((1, 2, 3))
df = my_df.filter(filter_func(a)(col('field1'))) \

【参考方案5】：

重申@zero323 上面提到的内容：我们也可以使用列表来做同样的事情（不仅是set），如下所示

from pyspark.sql.functions import col

df.where(col("v").isin(["foo", "bar"])).count()

【讨论】：

@zero323 是否在 LIKE 中而不是在 sparksql 中否定 is。

是的。你可以使用'~'