我想用类似 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 中手动写入值，因为它是从另一个作业中提取的。
在这种情况下我将如何过滤？
【问题讨论】：
标签： python sql apache-spark dataframe pyspark
只是一点点补充/更新：
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) ) )
【讨论】：
您传递给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 的所有细节。
【讨论】：
col，我可以轻松地解耦SQL 表达式和特定的DataFrame 对象。因此，例如，您可以保留有用的表达字典，并在需要时选择它们。使用显式 DF 对象，您必须将其放在函数中，并且它的组合并不好。
[(1,1), (1,2), (1,3)] 例如，其中一个是 aid，另一个是 bid。它必须类似于col(['aid', 'bid]).isin([(1,1), (1,2)])
您也可以对整数列执行此操作：
df_filtered = df.filter("field1 in (1,2,3)")
或者这个对于字符串列：
df_filtered = df.filter("field1 in ('a','b','c')")
【讨论】：
对我有用的一种稍微不同的方法是使用自定义过滤器功能进行过滤。
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'))) \
【讨论】：
重申@zero323 上面提到的内容：我们也可以使用列表来做同样的事情（不仅是set），如下所示
from pyspark.sql.functions import col
df.where(col("v").isin(["foo", "bar"])).count()
【讨论】：