Промежуточная сумма между двумя отметками времени в писпарке

Question

У меня есть данные в следующем формате:

+---------------------+----+----+---------+----------+
|      date_time      | id | cm | p_count |   bcm    |
+---------------------+----+----+---------+----------+
| 2018-02-01 04:38:00 | v1 | c1 |       1 |  null    |
| 2018-02-01 05:37:07 | v1 | c1 |       1 |  null    |
| 2018-02-01 11:19:38 | v1 | c1 |       1 |  null    |
| 2018-02-01 12:09:19 | v1 | c1 |       1 |  c1      |
| 2018-02-01 14:05:10 | v2 | c2 |       1 |  c2      |
+---------------------+----+----+---------+----------+

Мне нужно найти скользящую сумму столбца p_count между двумя date_time и разделом по идентификатору.

логика для окна start_date_time и end_date_time для скользящей суммы ниже:

start_date_time=min(date_time) group by (id,cm)

end_date_time= bcm == cm ? date_time : null

в этом случае start_date_time = 2018-02-01 04:38:00 и end_date_time = 2018-02-01 12:09:19.

Вывод должен выглядеть так:

+---------------------+----+----+---------+----------+-------------+
|      date_time      | id | cm | p_count |   bcm    | p_sum_count |
+---------------------+----+----+---------+----------+-------------+
| 2018-02-01 04:38:00 | v1 | c1 |       1 |  null    |1            |
| 2018-02-01 05:37:07 | v1 | c1 |       1 |  null    |2            |
| 2018-02-01 11:19:38 | v1 | c1 |       1 |  null    |3            |
| 2018-02-01 12:09:19 | v1 | c1 |       1 |  c1      |4            |
| 2018-02-01 14:05:10 | v2 | c2 |       1 |  c2      |1            |
+---------------------+----+----+---------+----------+-------------+

Answer 1

var input = sqlContext.createDataFrame(Seq(
            ("2018-02-01 04:38:00", "v1", "c1",1,null),
            ("2018-02-01 05:37:07", "v1", "c1",1,null),
            ("2018-02-01 11:19:38", "v1", "c1",1,null),
            ("2018-02-01 12:09:19", "v1", "c1",1,"c1"),
            ("2018-02-01 14:05:10", "v2", "c2",1,"c2")
            )).toDF("date_time","id","cm","p_count" ,"bcm")

    input.show()

Результаты:

+---------------------+----+----+---------+----------+-------------+
|      date_time      | id | cm | p_count |   bcm    | p_sum_count |
+---------------------+----+----+---------+----------+-------------+
| 2018-02-01 04:38:00 | v1 | c1 |       1 |  null    |1            |
| 2018-02-01 05:37:07 | v1 | c1 |       1 |  null    |2            |
| 2018-02-01 11:19:38 | v1 | c1 |       1 |  null    |3            |
| 2018-02-01 12:09:19 | v1 | c1 |       1 |  c1      |4            |
| 2018-02-01 14:05:10 | v2 | c2 |       1 |  c2      |1            |
+---------------------+----+----+---------+----------+-------------+

Следующий код:

        input.createOrReplaceTempView("input_Table");
        import org.apache.spark.sql.expressions.Window
        import org.apache.spark.sql.functions._

        //val results = spark.sqlContext.sql("SELECT sum(p_count) from input_Table tbl GROUP BY tbl.cm")

        val results = sqlContext.sql("select *, " +
          "SUM(p_count) over ( order by id  rows between unbounded preceding and current row ) cumulative_Sum " +
          "from input_Table ").show

Результаты:

+-------------------+---+---+-------+----+--------------+
|          date_time| id| cm|p_count| bcm|cumulative_Sum|
+-------------------+---+---+-------+----+--------------+
|2018-02-01 04:38:00| v1| c1|      1|null|             1|
|2018-02-01 05:37:07| v1| c1|      1|null|             2|
|2018-02-01 11:19:38| v1| c1|      1|null|             3|
|2018-02-01 12:09:19| v1| c1|      1|  c1|             4|
|2018-02-01 14:05:10| v2| c2|      1|  c2|             5|
+-------------------+---+---+-------+----+--------------+

Вам нужно сгруппировать по окнам и добавить свою логику, чтобы получить ожидаемые результаты

СТРОКИ МЕЖДУ НЕОГРАНИЧЕННОЙ ПЕРЕДАЧЕЙ И ТЕКУЩЕЙ СТРОКОЙ

Логически оконная агрегатная функция рассчитывается заново для каждой строки в пределах PARTITION на основе всех строк между начальной строкой и конечной строкой.

Начальная и конечная строки могут быть фиксированными или относительно текущей строки на основе следующих ключевых слов:

• UNBOUNDED PRECEDING, все строки перед текущей строкой -> исправлено
• UNBOUNDED FOLLOWING, все строки после текущей строки -> fixed
• x PRECEDING, x строк перед текущей строкой -> относительный
• y СЛЕДУЮЩИЙ, y строк после текущей строки -> относительный

Возможные виды расчета включают:

Как начальная, так и конечная строка являются фиксированными, окно состоит из всех строк раздела, например, Сумма группы, то есть совокупность плюс строки подробностей

Один конец фиксирован, другой относительно текущей строки, количество строк увеличивается или уменьшается, например, промежуточная сумма, оставшаяся сумма

Начальная и конечная строки относятся к текущей строке, количество строк в окне фиксировано, например, скользящее среднее по n строкам

Таким образом, SUM (x) OVER (ORDER BY col ROWS UNBOUNDED PRECEDING) приводит к накопленной сумме или промежуточной сумме

11 -> 11
 2 -> 11 +  2                = 13
 3 -> 13 +  3 (or 11+2+3)    = 16
44 -> 16 + 44 (or 11+2+3+44) = 60

Для чего в Teradata используется PROOEDING без строк?