Блог

это моя таблица:

  -------- ---- ---------- ---- 
| station|temp|     dateS|hour|
 -------- ---- ---------- ---- 
|Roma    | 2.2|2018-10-02|   1|
|Roma    | 1.5|2018-10-02|   2|
|Roma    | 1.4|2018-10-02|   3|
|Roma    | 1.4|2018-10-02|   4|
|Milano  | 0.6|2018-11-02|  12|
|Milano  | 1.0|2018-11-02|  13|
|Napoli  | 0.3|2018-12-02|  20|
|Napoli  | 0.0|2018-12-02|  21|
|Napoli  | 1.8|2018-12-02|   4|
|Napoli  | 2.0|2018-12-03|   5|
|Napoli  | 1.8|2018-12-03|   6|
 -------- ---- ---------- ---- 
  

Я хочу взять станцию, которая присутствует в записи в течение 4 часов подряд (4 часа подряд, очевидно, в один и тот же день).
Например, показывать Roma только потому, что в один и тот же день 4 часа подряд (2018-10-02)…

Я пытаюсь написать запрос, но ничего…

Я пробовал использовать функцию окна вывода. Я упорядочил фрейм данных, разделив его на станции и даты и упорядочив по часам, и вычислил разницу с предыдущим часом. Если мы рассматриваем в течение 4 часов подряд, в столбце разница должно быть три единицы одна за другой. Чтобы выяснить это, я собрал все различия на основе станции и дат и проверил, содержит ли он «1 1 1». Код для того же самого показан ниже. Я надеюсь, что это полезно.

 //Creating Test Data
  val df = Seq(("Roma",2.2,"2018-10-02",1 ) 
    , ("Roma",1.5,"2018-10-02",2 ) 
    , ("Roma",1.4,"2018-10-02",3 ) 
    , ("Roma",1.4,"2018-10-02",4 ) 
    , ("Milano",0.6,"2018-11-02",12 ) 
    , ("Milano",1.0,"2018-11-02",13 ) 
    , ("Napoli",0.3,"2018-12-02",20 ) 
    , ("Napoli",0.0,"2018-12-02",21 ) 
    , ("Napoli",1.8,"2018-12-02",4 ) 
    , ("Napoli",2.0,"2018-12-03",5 ) 
    , ("Napoli",1.8,"2018-12-03",6))
    .toDF("station", "temp", "dateS", "hour")

  val filterDF = df.withColumn("hour_lead", lead($"hour", 1)
                   .over(Window.partitionBy("station","dateS")
                               .orderBy(col("hour")))
                   .filter($"hour_lead".isNotNull)
                   .withColumn("hour_diff", $"hour_lead" - $"hour")
                   .groupBy("station","dateS")
          .agg(collect_list($"hour_diff".cast("string")).as("hour_diff_list"))
          .withColumn("hour_diff_list_str", 
                      concat(lit(" "), 
                      concat_ws(" ", $"hour_diff_list"), 
                      lit(" ")))
          .filter($"hour_diff_list_str".contains(" 1 1 1 "))

  filterDF.show(false)
   ------- ---------- -------------- ------------------ 
  |station|dateS     |hour_diff_list|hour_diff_list_str|
   ------- ---------- -------------- ------------------ 
  |Roma   |2018-10-02|[1, 1, 1]     | 1 1 1            |
   ------- ---------- -------------- ------------------