PySpark 子查询:不允许访问外部查询列

Question

我正在尝试将SQL查询重新编写到PySpark中。下面是 SQL 查询：

SELECT 
cs.Environment, 
cs.AccountCode,  
MIN(cs.StartDate) AS StartDate,         
MIN(cs.FinalDate) AS FinalDate, 
(
SELECT TOP 1 ItemCode 
FROM [dbo].[Contracts] 
WHERE 
Environment = cs.Environment 
AND AccountCode = cs.AccountCode 
AND ContractType = 'C'      
AND LinePackage = 1             
AND InflowOutflow = 'Inflow'    
AND EventDate <= GETDATE()      
ORDER BY EventDate 
) AS Package
FROM [dbo].[Contracts] cs
WHERE 
cs.ContractType = 'C'
AND cs.LinePackage  = 1
GROUP BY 
cs.Environment, 
cs.AccountCode

我的 PySpark 代码是这样的：

df = spark.sql(
"""select cs.environment, cs.accountcode, 
min(cs.startdatets) as startdate, min(cs.finaldatets) as finaldate,
(select a.itemcode 
from firstcomm as a 
where a.environment = cs.environment and a.accountcode = cs.accountcode and a.contracttype = 'c' and a.eventdate <= current_date() 
order by a.eventdate limit 1) as package 
from firstcomm cs where cs.contracttype = 'c' and cs.linepackage  = 1 
group by cs.environment, cs.accountcode""")

但我不断收到此错误：

AnalysisException: Accessing outer query column is not allowed in:
LocalLimit 1
+- Project [itemcode#3641]
+- Sort [eventdate#3629 ASC NULLS FIRST], true
+- Project [itemcode#3641, eventdate#3629]
+- Filter ((((environment#3628 = outer(environment#3628)) && (accountcode#3622 = outer(accountcode#3622))) && (contracttype#3626 = c)) && (((linepackage#3644 = 1) && (inflowoutflow#3637 = inflow)) && (eventdate#3629 <= current_date(Some(Zulu)))))
+- SubqueryAlias a

顺便说一句，我正在使用Spark 2.2.1，我相信它支持子查询

有什么想法可以解决这个问题吗？或者我应该如何重写查询以获得所需的结果？

Answer 1

你习惯使用 pyspark 数据帧 API 吗？

import pyspark.sql.functions as F
from pyspark.sql.window import Window
# get the top item code for each environment/account code
package_df = contracts.filter((F.col("ContractType") == 'C') & (F.col("LinePackage") == 1) & (F.col("InflowOutflow") == "Inflow") & (F.col("EventDate") <= GETDATE())).orderBy("EventDate")
package_df = package_df.withColumn("row_number", F.row_number().over(Window.partitionBy("Environment", "AccountCode").orderBy(F.col("ItemCode").desc())
package_df = package_df.filter(F.col("row_number") == 1).drop("row_number")
# aggregate over your contracts table and join the top item code
contracts_df = contracts.filter((F.col("ContractType") == 'C') & (F.col("AccountCode") == 1))
contracts_df = contracts_df.groupBy("Environment", "AccountCode").agg(F.min("StartDate").alias("StartDate"), F.min("FinalDate").alias("FinalDate"))
# join to get the top item code
output_df = contracts_df.join(package_df, ["Environment", "AccountCode"])

我使用窗口函数获取每个键的顶部项目代码，然后将其联接到聚合的原始数据帧中。