Spark Scala - минимальная разница во временных метках

Question

Я пытаюсь выяснить за 24-часовой период времени, какова минимальная и максимальная температура в заданном c 24-часовом окне, а затем найти день с наибольшей разницей. Как только у меня появляется эта самая большая разница, я пытаюсь определить кратчайший период времени, в котором эта разница возникла.

Например, пример моих обработанных данных выглядит так: Текущий вывод из кода ниже:

+------------+-------------------+---------+-------+-------+
|TemperatureF|               Date|timestamp|MinTemp|MaxTemp|
+------------+-------------------+---------+-------+-------+
|        28.0| 01/01/2000 6:53 AM|946709580|   28.0|   37.4|
|        28.0| 01/01/2000 7:53 AM|946713180|   28.0|   37.4|
|        28.0| 01/01/2000 8:53 AM|946716780|   28.0|   37.4|
|        30.2|01/01/2000 10:24 PM|946765440|   30.2|   37.4|
|        30.9|01/01/2000 10:53 PM|946767180|   30.9|   37.4|
|        37.4| 01/02/2000 4:39 AM|946787940|   28.0|   37.4|
|        36.0| 01/02/2000 4:53 AM|946788780|   28.0|   36.0|
|        36.0| 01/02/2000 5:53 AM|946792380|   28.0|   36.0|
+------------+-------------------+---------+-------+-------+

Если я посмотрю на 3 первые строки, где MinTemp равен 28, а MaxTemp равен 37,4. Так как я смотрю на разницу между этими 2 (9,4) за 24-часовой период (с 1/1/2000 6:53 до 1/2/2000 6:53 AM). MaxTemp происходит в 01.02.2000 в 4:39. Учитывая это, я бы хотел, чтобы мой ответ был:

+-----------------+-------------------+----------+
|Start Time       |            EndTime|difference|
+-----------------+-------------------+----------+
|01/01/2000 8:53AM|  01/02/2000 4:39AM|9.4       |
+-----------------+-------------------+----------+

Это потому, что это самый короткий период времени для достижения этой разницы в температуре.

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

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

+-----------------+-----------+-----------------+-----------+-------+-------+
StartTime         |Temperature|EndTime.         |Temperature|MinTemp|MaxTemp|
+-----------------+-----------+-----------------+-----------+-------+-------+
01/01/2000 6:53AM | 28.0      |01/02/2000 4:39AM| 37.4      | 28.0  | 37.4  |
01/01/2000 7:53AM | 28.0      |01/02/2000 4:39AM| 37.4      | 28.0  | 37.4  |
01/01/2000 8:53AM | 28.0      |01/02/2000 4:39AM| 37.4      | 28.0  | 37.4  |
+-----------------+-----------+-----------------+-----------+-------+-------+

Отсюда я бы рассчитал минимальную разницу во времени и разницу в температуре.

Есть какие-нибудь мысли о том, как это сделать?

Текущий код:

val data = osh.select(col("TemperatureF"), concat(format_string("%02d",col("Month")),lit("/"),format_string("%02d",col("Day")),lit("/"),col("Year"),lit(" "),col("TimeCST")).as("Date")).filter(col("TemperatureF") > -9999)

val oshdata = data.withColumn("timestamp",unix_timestamp(to_timestamp(col("Date"),"MM/dd/yyyy hh:mm a")))

import org.apache.spark.sql.expressions._
val myWindow = Window.orderBy("timestamp").rangeBetween(Window.currentRow, 86400)
val myData = oshdata.withColumn("MinTemp", min(col("TemperatureF")).over(myWindow))
  .withColumn("MaxTemp",max(col("TemperatureF")).over(myWindow))


myData.createOrReplaceTempView("oshView")
spark.sqlContext.sql("Select * from oshView where TemperatureF == MinTemp or TemperatureF == MaxTemp").show(25)

Answer 1

Я импортировал данные из вашей публикации и написал следующее SQL, чтобы помочь решить ваш вопрос (основываясь на моем понимании из вашего вопроса ...); Тем не менее, я надеюсь, что это помогает.

    df = spark_sql.read
        .format("csv")
        .option("header", "true")
        .option("delimiter", ",") 
        .load("/path/to/minimum-difference-in-timestamps.txt")

    df.registerTempTable("TABLE1")
    spark_sql.sql(""" CACHE TABLE View1 AS 
                      SELECT *  FROM TABLE1 ORDER BY TemperatureF DESC LIMIT 1""")

---

    spark_sql.sql("""  CACHE TABLE View2 AS 
                       SELECT A.Date StartTime,
                              B.Date EndTime, 
                              CAST(B.TemperatureF - A.TemperatureF AS FLOAT) difference ,
                              A.TemperatureF,
                              A.MinTemp,
                              A.MaxTemp,
                              B.timestamp END_TIME_TS
                       FROM (
                                SELECT * FROM TABLE1 WHERE 
                                timestamp < (SELECT timestamp FROM View1) 
                                ORDER BY TemperatureF ASC, timestamp DESC LIMIT 1 
                            ) AS A JOIN 
                              View1 AS B ON A.MaxTemp == B.TemperatureF""")

---

    spark_sql.sql(""" SELECT StartTime,EndTime,difference FROM View2""").show()
    spark_sql.sql(""" SELECT A.Date StartTime, 
                             A.TemperatureF Temperature,
                             B.EndTime ,
                             B.TemperatureF Temperature_END,
                             B.MinTemp, 
                             B.MaxTemp
                     FROM  TABLE1 AS A JOIN
                           View2 AS B 
                           ON A.MaxTemp =B.MaxTemp 
                              AND 
                              A.MinTemp == B.MinTemp
                              AND 
                              A.timestamp < B.END_TIME_TS
     """).show()


+-----------------+-----------------+----------+
|        StartTime|          EndTime|difference|
+-----------------+-----------------+----------+
|01/01/2000 8:53AM|01/02/2000 4:39AM|       9.4|
+-----------------+-----------------+----------+

+-----------------+-----------+-----------------+---------------+-------+-------+
|        StartTime|Temperature|          EndTime|Temperature_END|MinTemp|MaxTemp|
+-----------------+-----------+-----------------+---------------+-------+-------+
|01/01/2000 6:53AM|       28.0|01/02/2000 4:39AM|           28.0|   28.0|   37.4|
|01/01/2000 7:53AM|       28.0|01/02/2000 4:39AM|           28.0|   28.0|   37.4|
|01/01/2000 8:53AM|       28.0|01/02/2000 4:39AM|           28.0|   28.0|   37.4|
+-----------------+-----------+-----------------+---------------+-------+-------+

---

минимальная разница во времени timestamps.txt:

TemperatureF,Date,timestamp,MinTemp,MaxTemp
28.0,01/01/2000 6:53AM,946709580,28.0,37.4
28.0,01/01/2000 7:53AM,946713180,28.0,37.4
28.0,01/01/2000 8:53AM,946716780,28.0,37.4
30.2,01/01/2000 10:24PM,946765440,30.2,37.4
30.9,01/01/2000 10:53PM,946767180,30.9,37.4
37.4,01/02/2000 4:39AM,946787940,28.0,37.4
36.0,01/02/2000 4:53AM,946788780,28.0,36.0
36.0,01/02/2000 5:53AM,946792380,28.0,36.0