Блог

У меня есть DataFrame следующий простой schema :

 root
 |-- amount: double (nullable = true)
 |-- Date: timestamp (nullable = true)
  

Я пытался увидеть sum количество в день и в час, например:

  --- -------- --------  ...  -------- 
|day|       0|       1|     |      23|
 --- -------- --------  ...  -------- 
|148|   306.0|   106.0|     |     0.0|
|243|  1906.0|    50.0|     |     1.0|
| 31|   866.0|   100.0|     |     0.0|
 --- -------- --------  ...  -------- 
  

Ну, сначала я добавил столбец hour , а затем сгруппировал по дням и развернул по часам. Однако я получил исключение, которое, возможно, связано с отсутствием продаж в течение нескольких часов. Это то, что я пытаюсь исправить, но я не понял, как.

 (df.withColumn("hour", hour("date"))
    .groupBy(dayofyear("date").alias("day"))
    .pivot("hour")
    .sum("amount").show())
  

Выдержка из исключения.

Исключение AnalysisException: u’решенный атрибут (ы) дата # 3972 отсутствует в день # 5367, час # 5354, сумма (сумма) # 5437 в операторе!Агрегировать [dayofyear(приведение (дата #3972 в качестве даты))], [dayofyear(приведение (дата#3972 в качестве даты)) КАК день # 5367, pivotfirst(час # 5354, сумма (сумма)#5437, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20,21, 22, 23, 0, 0) КАК __сводная_сумма(сумма) КАК сумма(сумма)#5487];’

Проблема в нерешенном day столбце. Вы можете создать groupBy предложение it outside для решения этой проблемы:

 df = (sc
         .parallelize([
             (1.0, "2016-03-30 01:00:00"), (30.2, "2015-01-02 03:00:02")])
        .toDF(["amount", "Date"])
        .withColumn("Date", col("Date").cast("timestamp"))
        .withColumn("hour", hour("date")))

with_day = df.withColumn("day", dayofyear("Date"))
with_day.groupBy("day").pivot("hour", range(0, 24)).sum("amount")
  

values аргумент for pivot необязателен, но рекомендуется.