Filling gaps in timeseries Spark

by

If input DataFrame has following structure:

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

Scala

Determine min / max:

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

Set step, for example for 15 minutes:

val step: Long = 15 * 60

Generate reference range:

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

Join and fill the gaps:

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

Python

Similarly in 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")

More Related Contents:

  1. Spark 2.0 Dataset vs DataFrame
  2. Spark – load CSV file as DataFrame?
  3. How to define and use a User-Defined Aggregate Function in Spark SQL?
  4. How to define partitioning of DataFrame?
  5. Automatically and Elegantly flatten DataFrame in Spark SQL
  6. Spark extracting values from a Row
  7. Caused by: java.lang.NullPointerException at org.apache.spark.sql.Dataset
  8. SparkSQL: How to deal with null values in user defined function?
  9. How to aggregate values into collection after groupBy?
  10. Dropping a nested column from Spark DataFrame
  11. How to read records in JSON format from Kafka using Structured Streaming?
  12. Spark Dataframe :How to add a index Column : Aka Distributed Data Index
  13. Defining a UDF that accepts an Array of objects in a Spark DataFrame?
  14. Processing multiple files as independent RDD’s in parallel
  15. MatchError while accessing vector column in Spark 2.0
  16. Spark / Scala: forward fill with last observation
  17. About how to add a new column to an existing DataFrame with random values in Scala
  18. How to compare two dataframe and print columns that are different in scala
  19. Perform a typed join in Scala with Spark Datasets
  20. Apache Spark: Get number of records per partition
  21. Renaming column names of a DataFrame in Spark Scala
  22. DataFrame-ified zipWithIndex
  23. Derive multiple columns from a single column in a Spark DataFrame
  24. How to save a spark DataFrame as csv on disk?
  25. What is going wrong with `unionAll` of Spark `DataFrame`?
  26. How to sort by column in descending order in Spark SQL?
  27. How to get keys and values from MapType column in SparkSQL DataFrame
  28. Why is join not possible after show operator?
  29. How to create DataFrame from Scala’s List of Iterables?
  30. How to create a Dataset of Maps?

Leave a Comment