Pyspark - время раунда, представленное как целое число с точностью до четверти часа (15 минут) - PullRequest
Новая
       12

Pyspark - время раунда, представленное как целое число с точностью до четверти часа (15 минут)

0 голосов
/ 15 мая 2018

Я работаю над датафреймом в Pyspark. Один столбец состоит из целочисленного значения, соответствующего времени: 

data.select('TIME').show(4)
+------------------+
|TIME              |
+------------------+
|               925|
|              2205|
|              2205|
|              2205|
+------------------+

Я бы хотел округлить это время до ближайшего значения с шагом по времени 15 минут, чтобы получить: 

+------------------+
|TIME_15_MIN_STEP  |
+------------------+
|               930|
|              2200|
|              2200|
|              2200|
+------------------+

Кто-нибудь знает, как это сделать?

Большое спасибо !!

1 Ответ

0 голосов
/ 21 мая 2018

Один из способов сделать это без использования udf - сначала преобразовать ваш целочисленный столбец в фиктивную временную метку, а затем сделать почти то же самое, что описано в моем ответе в .аналогичный вопрос .Наконец, преобразуйте результат обратно в целое число желаемого формата.

Более подробные данные примера

Я создал пример с еще большей вариативностью, чтобы продемонстрировать, что этот метод работает

data = sqlCtx.createDataFrame([(925,), (2205,), (2210,), (2242,), (2255,)], ["TIME"])
data.show()
#+----+
#|TIME|
#+----+
#| 925|
#|2205|
#|2210|
#|2242|
#|2255|
#+----+

Преобразование целочисленного столбца в фиктивную метку времени

Чтобы преобразовать целочисленный час-минутный столбец в метку времени, мы сначала используем pyspark.sql.functions.format_string()От до добавьте начальные нули к соответствующему времени.Затем мы объединяем фиктивную дату (я использовал "2018-01-01") с преобразованным временем и добавляем ":00" в конце (в секундах). 

data = data.withColumn("time_string", f.format_string("%04d", f.col("TIME")))\
    .withColumn(
        "time_string",
        f.concat_ws(
            ":",
            f.array(
                [
                    f.substring(
                        "time_string",
                        1,
                        2
                    ),
                    f.substring(
                        "time_string",
                        3,
                        2
                    ),
                    f.lit("00")
                ]
            )
        )
    )\
    .withColumn("time_string", f.concat(f.lit("2018-01-01 "), f.col("time_string")))

data.show()
#+----+-------------------+
#|TIME|        time_string|
#+----+-------------------+
#| 925|2018-01-01 09:25:00|
#|2205|2018-01-01 22:05:00|
#|2210|2018-01-01 22:10:00|
#|2242|2018-01-01 22:42:00|
#|2255|2018-01-01 22:55:00|
#+----+-------------------+

Вычисляем, сколько минут нужно для смещенияотметка времени

Используйте pyspark.sql.functions.minute() для получения минуты из фиктивной отметки времени.Мы делим на 15, округляем, а затем умножаем на 15, чтобы получить «новую» минуту.(Эта логика более подробно объясняется в связанном ответе .) 

data = data.withColumn("minute", f.minute("time_string"))\
    .withColumn("new_minute", f.round(f.col("minute")/15)*15)\
    .withColumn("minute_add", f.col("new_minute") - f.col("minute"))\

data.show()
#+----+-------------------+------+----------+----------+
#|TIME|        time_string|minute|new_minute|minute_add|
#+----+-------------------+------+----------+----------+
#| 925|2018-01-01 09:25:00|    25|      30.0|       5.0|
#|2205|2018-01-01 22:05:00|     5|       0.0|      -5.0|
#|2210|2018-01-01 22:10:00|    10|      15.0|       5.0|
#|2242|2018-01-01 22:42:00|    42|      45.0|       3.0|
#|2255|2018-01-01 22:55:00|    55|      60.0|       5.0|
#+----+-------------------+------+----------+----------+

Добавить смещение в секундах, преобразовать обратно в целое число

Умножьте столбец minute_add на 60, чтобы получить смещение в секундах.Добавьте это к time_string, чтобы получить «новое» время. 

data = data.withColumn(
        "new_time",
        f.from_unixtime(f.unix_timestamp("time_string") + f.col("minute_add")*60)
    )\
    .withColumn(
        "NEW_TIME",
        f.format_string("%02d%02d", f.hour("new_time"), f.minute("new_time")).cast("int")
    )
data.select("TIME", "NEW_TIME").show()
#+----+--------+
#|TIME|NEW_TIME|
#+----+--------+
#| 925|     930|
#|2205|    2200|
#|2210|    2215|
#|2242|    2245|
#|2255|    2300|
#+----+--------+
...