Коррелированный столбец подзапроса в SPARK SQL не допускается как часть предиката неравенства

Question

Я пытаюсь написать подзапрос в предложении where, как показано ниже.Но я получаю "Коррелированный столбец недопустим в предикате неравенства:"

SELECT *, 
   holidays 
      FROM   ( 
          SELECT *, 
                 s.holidays, 
                 s.entity 
          FROM   transit_t tt 
          WHERE  ( 
                        SELECT Count(thedate) AS holidays 
                        FROM   fact_ent_rt 
                        WHERE  entity=tt.awborigin 
                        AND    ( 
                                      Substring(thedate,1,10)) BETWEEN (Substring(awbpickupdate,1,10)) AND    (
                                      Substring(deliverydate,1,10)) 
                        AND    ( 
                                      nholidayflag = true 
                               OR     weekendflag = true))) s

Любые проблемы с этим запросом.потому что я думал, что spark> 2.0 поддерживает подзапросы в предложении where.Мы ценим любые предложения.Спасибо

На входе будет дата получения и дата доставки из транзитной таблицы.Нам нужно выяснить, есть ли выходные дни, попадающие между этими датами (эти данные доступны в fact_ent_rt), и рассчитать количество выходных.

Вывод, который я получаю, - это pyspark.sql.utils.AnalysisException:u "Коррелированный столбец недопустим в предикате неравенства: \ nAggregate

Пример ввода:

Вход 1: + --------- ++ ------------------- + ------------------- +

|AWBOrigin||      AWBPickupDate|       DeliveryDate|

+---------++-------------------+-------------------+

|      LON||2018-09-01 08:52:00|2018-09-12 13:57:00|
|      DHA||2018-09-04 11:47:00|2018-09-08 07:30:00|
|      NIC||2009-01-01 01:47:00|2009-01-09 11:37:00
+---------+-----------+-----------------------------

Ввод 2 (fact_ent)

------+-------------------+-----------+------------

 Entity|            TheDate|WeekendFlag|NHolidayFlag

 ------+-------------------+-----------+------------

NIC|2009-01-01 00:00:00|      False|       False
NIC|2009-01-02 00:00:00|      False|       False
NIC|2009-01-03 00:00:00|       True|       False
NIC|2009-01-04 00:00:00|       True|       False
NIC|2009-01-05 00:00:00|      False|       False
NIC|2009-01-06 00:00:00|      False|       False
NIC|2009-01-07 00:00:00|      False|       False
NIC|2009-01-08 00:00:00|      False|       False
NIC|2009-01-09 00:00:00|      False|       False
NIC|2009-01-10 00:00:00|       True|       False
NIC|2009-01-11 00:00:00|       True|       False
NIC|2009-01-12 00:00:00|      False|       False
NIC|2009-01-13 00:00:00|      False|       False
NIC|2009-01-14 00:00:00|      False|       False
NIC|2009-01-15 00:00:00|      False|       False
NIC|2009-01-16 00:00:00|      False|       False
NIC|2009-01-17 00:00:00|       True|       False
NIC|2009-01-18 00:00:00|       True|       False
NIC|2009-01-19 00:00:00|      False|       False
NIC|2009-01-20 00:00:00|      False|       False
------+-------------------+-----------+------------

expectede Вывод

 +---------++-------------------+-------------------+

|AWBOrigin||      AWBPickupDate|       DeliveryDate| Holidays

+---------++-------------------+-------------------+

|      LON||2018-09-01 08:52:00|2018-09-12 13:57:00|  NA
|      DHA||2018-09-04 11:47:00|2018-09-08 07:30:00|  NA
|      NIC||2009-01-01 01:47:00|2009-01-09 11:37:00|  2
+---------+-----------+-----------------------------

Answer 1

Я сделал это со SCALA, так что вам нужно будет конвертировать, но я думаю, что это намного проще. Я добавил ключ и сделал на ключевом уровне, вы можете адаптировать и усвоить это. Но принцип намного проще. Не требуется коррелированных подзапросов. Просто реляционное исчисление. Используемый номер для дат и т. Д.

// SCALA 
// Slightly ambiguous on hols vs. weekend, as you stated treated as 1

import spark.implicits._ 
import org.apache.spark.sql.functions._

val dfE = Seq( 
              ("NIC", 1, false, false),
              ("NIC", 2, false, false),
              ("NIC", 3, true, false),
              ("NIC", 4, true, true),
              ("NIC", 5, false, false),
              ("NIC", 6, false, false),
              ("XYZ", 1, false, true)
              ).toDF("e","d","w", "h")
 //dfE.show(false)

 val dfE2 = dfE.withColumn("wh", when ($"w" or $"h", 1) otherwise (0)).drop("w").drop("h")
 //dfE2.show()

//Assuming more dfD's can exist
val dfD = Seq( 
              ("NIC", 1, 4, "k1"),
              ("NIC", 2, 3, "k2"),
              ("NIC", 1, 1, "k3"),
              ("NIC", 7, 10, "k4")
              ).toDF("e","pd","dd", "k")
//dfD.show(false)

dfE2.createOrReplaceTempView("E2")
dfD.createOrReplaceTempView("D1")

// This done per record, if over identical keys, then strip k and aggr otherwise, I added k for checking each entry
// Point is it is far easier. Key means synthetic grouping by.

val q=sqlContext.sql(""" SELECT d1.k, d1.e, d1.pd, d1.dd, sum(e2.wh) 
                       FROM D1, E2
                      WHERE D1.e = E2.e 
                        AND E2.d >= D1.pd
                        AND E2.d <= D1.dd
                    GROUP BY d1.k, d1.e, d1.pd, d1.dd   
                    ORDER BY d1.k, d1.e, d1.pd, d1.dd
                     """)
q.show

возвращается:

 +---+---+---+---+-------+
 |  k|  e| pd| dd|sum(wh)|
 +---+---+---+---+-------+
 | k1|NIC|  1|  4|      2|
 | k2|NIC|  2|  3|      1|
 | k3|NIC|  1|  1|      0|
 +---+---+---+---+-------+

Я думаю, что простое улучшение производительности может быть сделано. На самом деле никаких взаимосвязанных вещей не требуется.

Можно использовать AND E2.d МЕЖДУ D1.pd И D1.dd, если хотите.