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

Question

我想用类似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]失败:``('''',但是找到了标识符

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

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

Answer 1

传递给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的所有详细信息.

Answer 2

重申@ zero323上面提到的内容:我们也可以使用列表(不仅set如下)做同样的事情

from pyspark.sql.functions import col

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

Answer 3

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

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

或者这对于字符串列：

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

Answer 4

只需一点补充/更新：

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

如果要过滤数据框“ df”，以便要基于仅从choice_list取值的列“ v”保留行，则

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