PySpark SubQuery: доступ к внешнему столбцу запроса не разрешен - PullRequest
Новая
       7

PySpark SubQuery: доступ к внешнему столбцу запроса не разрешен

0 голосов
/ 16 октября 2018

Я пытаюсь переписать запрос 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, который, как мне кажется, поддерживает подзапросы

Есть идеи, как решить эту проблему?или как мне переписать запрос, чтобы получить желаемый результат?

1 Ответ

0 голосов
/ 16 октября 2018

Вам удобно использовать API данных pyspark? 

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"])

Я использовал оконные функции, чтобы получить код верхнего элемента для ключа, а затем присоединил его к вашему агрегированному исходному кадру данных.

...