scala - Rellenar huecos en series temporales Spark

apache-spark apache-spark-sql

Question

Tengo un problema con los datos de series temporales. Debido a fallas de alimentación faltan algunas marcas de tiempo en el conjunto de datos. Necesito llenar estos vacíos agregando filas, y después de eso, puedo interpolar los valores faltantes.

Datos de entrada:

periodstart                usage
---------------------------------
2015-09-11 02:15           23000   
2015-09-11 03:15           23344   
2015-09-11 03:30           23283  
2015-09-11 03:45           23786   
2015-09-11 04:00           25039

Salida deseada:

periodstart                usage
---------------------------------
2015-09-11 02:15           23000   
2015-09-11 02:30           0   
2015-09-11 02:45           0   
2015-09-11 03:00           0   
2015-09-11 03:15           23344   
2015-09-11 03:30           23283   
2015-09-11 03:45           23786   
2015-09-11 04:00           25039  

Ahora lo he solucionado con un ciclo while dentro de una función foreach de conjunto de datos. El problema es que tengo que recopilar el conjunto de datos primero en el controlador antes de poder hacer un ciclo while. Así que esa no es la forma correcta para Spark.

¿Alguien puede darme una mejor solución?

Este es mi código:

MissingMeasurementsDS.collect().foreach(row => {
  // empty list for new generated measurements
  val output = ListBuffer.empty[Measurement]
  // Missing measurements
  val missingMeasurements = row.getAs[Int]("missingmeasurements")
  val lastTimestamp = row.getAs[Timestamp]("previousperiodstart")
  //Generate missing timestamps
  var i = 1
  while (i <= missingMeasurements) {
    //Increment timestamp with 15 minutes (900000 milliseconds)
    val newTimestamp = lastTimestamp.getTime + (900000 * i)
    output += Measurement(new Timestamp(newTimestamp), 0))
    i += 1
  }
  //Join interpolated measurements with correct measurements
  completeMeasurementsDS.join(output.toDS())
})
completeMeasurementsDS.show()
println("OutputDF count = " + completeMeasurementsDS.count())

Answer 1

Si el DataFrame entrada tiene la siguiente estructura:

root
 |-- periodstart: timestamp (nullable = true)
 |-- usage: long (nullable = true)

Scala

Determinar min / max:

val (minp, maxp) = df
  .select(min($"periodstart").cast("bigint"), max($"periodstart".cast("bigint")))
  .as[(Long, Long)]
  .first

Establezca el paso, por ejemplo, durante 15 minutos:

val step: Long = 15 * 60

Generar rango de referencia:

val reference = spark
  .range((minp / step) * step, ((maxp / step) + 1) * step, step)
  .select($"id".cast("timestamp").alias("periodstart"))

Únete y llena los vacíos:

reference.join(df, Seq("periodstart"), "leftouter").na.fill(0, Seq("usage"))

Pitón

Del mismo modo en PySpark:

from pyspark.sql.functions import col, min as min_, max as max_

step = 15 * 60

minp, maxp = df.select(
    min_("periodstart").cast("long"), max_("periodstart").cast("long")
).first()

reference = spark.range(
    (minp / step) * step, ((maxp / step) + 1) * step, step
).select(col("id").cast("timestamp").alias("periodstart"))

reference.join(df, ["periodstart"], "leftouter")