使用where或filter语句在pyspark中运行子查询

Question

我正在尝试在pyspark中运行子查询。我发现可以使用SQL语句。但是，使用“where”或“filter”操作是否有任何内在的支持？

考虑测试数据框架：

from pyspark.sql import SparkSession
sqlContext = SparkSession.builder.appName('test').enableHiveSupport().getOrCreate() 
tst = sqlContext.createDataFrame([(1,2),(4,3),(1,4),(1,5),(1,6)],schema=['sample','time'])
tst_sub = sqlContext.createDataFrame([(1,2),(4,3),(1,4)],schema=['sample','time'])
#%% using where to query the df
tst.where(F.col('time')>4).show()
+------+----+
|sample|time|
+------+----+
|     1|   5|
|     1|   6|
+------+----+

在这里，您可以看到where函数工作正常。当我尝试使用子查询执行相同操作时，如下所示：

#%% using where with subquery
tst.where(F.col('time')>F.max(tst_sub.select('time'))).show()

我得到这个错误：

属性错误回溯（最近一次调用）在----

/opt/cloudera/parcels/CDH-6.3.4-1。cdh6.3.4。p4744.12781922/lib/spark/python/pyspark/sql/functions。py in（col）42 def（col）：43 sc=SparkContext_active\u spark\u上下文---

/opt/cloudera/parcels/CDH-6.3.4-1。cdh6.3.4。p4744.12781922/lib/spark/python/lib/py4j-0.10.7-src。zip/py4j/java\u网关。py in call（self，*args）1246 1247 def call（self，*args）：-

/opt/cloudera/parcels/CDH-6.3.4-1。cdh6.3.4。p4744.12781922/lib/spark/python/lib/py4j-0.10.7-src。zip/py4j/java\u网关。py in \u build\u args（self，*args）1216 1217 args\u命令=“”。加入（-

/opt/cloudera/parcels/CDH-6.3.4-1。cdh6.3.4。p4744.12781922/lib/spark/python/lib/py4j-0.10.7-src。zip/py4j/java\u网关。py in（.0）1216 1217 args\u命令=“”。加入（-

/opt/cloudera/parcels/CDH-6.3.4-1。cdh6.3.4。p4744.12781922/lib/spark/python/lib/py4j-0.10.7-src。zip/py4j/协议。py in get\u command\u part（参数，python\u proxy\u pool）296 command\u part=“；”接口297其他：--

/opt/cloudera/parcels/CDH-6.3.4-1.cdh6.3.4. p4744.12781922 /lib/spark/python/pyspark/sql/dataframe.py在getattr（自己，名称）1298如果名称不在self.columns： 1299引发属性错误（-

属性错误：'DataFrame'对象没有属性'_get_object_id'

当我将数据帧注册为表并执行sql查询时，它工作正常：

tst.createOrReplaceTempView("tst")
tst_sub.createOrReplaceTempView("tst_sub")
sqlContext.sql("SELECT * FROM tst WHERE time>(SELECT(max(time)) FROM tst_sub)").show()

是否有任何方法可以直接使用filter、where或任何其他方法在pyspark中对数据帧执行子查询？

Answer 1

在将最大时间放入过滤器之前，需要将其收集到Python中的数字变量中：

tst.where(F.col('time') > tst_sub.select(F.max('time')).head()[0]).show()
+------+----+
|sample|time|
+------+----+
|     1|   5|
|     1|   6|
+------+----+