У меня есть данные потоковой метки времени и я хочу рассчитать некоторые статистические данные за 10-минутные интервалы времени:
out_df = stream\
.groupby(F.window(st.timestamp, "600 seconds"))\
.agg(F.sum(), F.agv())\
...
Однако я обнаружил, что если в конкретном интервале нет данных в потоке, то запись для этого интервала не создается.
Есть ли способ заполнить пропущенные строки интервалами, для которых нет данных?
Хотя мой конкретный вопрос касается структурированной потоковой передачи, ниже приведен простой способ воспроизведения на статическом фрейме данных:
>>> df = spark.createDataFrame([("2016-03-11 09:00:07", 1),("2016-03-11 09:00:08", 1),("2016-03-11 09:02:07", 1)]).toDF("date", "val")
>>>
>>> w = df.groupBy(F.window("date", "60 seconds")).agg(F.sum("val").alias("sum"))
>>> from pyspark.sql import functions as F
>>> w.select(w.window.start.cast("string").alias("start"),w.window.end.cast("string").alias("end"), "sum").collect()
[Row(start='2016-03-11 09:02:00', end='2016-03-11 09:03:00', sum=1), Row(start='2016-03-11 09:00:00', end='2016-03-11 09:01:00', sum=2)]
как вы можете видеть, строка с интервалом 09: 01: 00-09: 02: 00 отсутствует, и я хотел бы иметь ее, возможно, со значениями NaN или 0 с.
Это Spark 2.3.1 (HDP 3.0).