使用 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()

我收到此错误：

AttributeError Traceback（最近调用 最后）在 ----> 1 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 在 _(col) 42定义_（col）： 43 sc = SparkContext._active_spark_context ---> 44 jc = getattr(sc._jvm.functions, name)(col._jc if isinstance(col, Column) else col) 45返回列（jc） 46 _.姓名 = 姓名

/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_gateway.py 在 call(self, *args) 1246 1247 def call(self, *参数）： -> 1248 args_command, temp_args = self._build_args(*args) 1249 1250 command = proto.CALL_COMMAND_NAME +\

/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_gateway.py 在 _build_args(self, *args) 1216 1217 args_command = ““。加入（ -> 1218 [get_command_part(arg, self.pool) for arg in new_args]) 1219 1220 return args_command, temp_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_gateway.py 在 (.0) 1216 1217 args_command = "".join( -> 1218 [get_command_part(arg, self.pool) for arg in new_args]) 1219 1220 return args_command, temp_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/protocol.py get_command_part（参数，python_proxy_pool） 第296章+ 界面 297 其他： --> 298 command_part = REFERENCE_TYPE + parameter._get_object_id() 299 300 command_part += "\n"

/opt/cloudera/parcels/CDH-6.3.4-1.cdh6.3.4.p4744.12781922/lib/spark/python/pyspark/sql/dataframe.py 在 getattr(self, name) 1298 如果名称不在 self.columns：1299 引发 AttributeError（ -> 1300 "'%s' 对象没有属性 '%s'" % (self.class.name, name)) 1301 jc = self._jdf.apply(name) 1302 返回列(jc)

AttributeError: '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()

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

Answer 1

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

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