PySpark: how to resample frequencies

by

Question: How to do that on a Spark Dataframe in an efficient way?

Spark DataFrame is simply not a good choice for an operation like this one. In general SQL primitives won’t be expressive enough and PySpark DataFrame doesn’t provide low level access required to implement it.

While re-sampling can be easily represented using epoch / timestamp arithmetics. With data like this:

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

df = (spark  
    .createDataFrame([
        ("2012-06-13", 0.694), ("2012-06-20", -2.669), ("2012-06-27", 0.245)],   
        ["ts", "val"])        
   .withColumn("ts", col("ts").cast("date").cast("timestamp")))

we can re-sample input:

day = 60 * 60 * 24
epoch = (col("ts").cast("bigint") / day).cast("bigint") * day

with_epoch = df.withColumn("epoch", epoch)

min_epoch, max_epoch = with_epoch.select(min_("epoch"), max_("epoch")).first()

and join with reference:

# Reference range 
ref = spark.range(
    min_epoch, max_epoch + 1, day
).toDF("epoch")

(ref
    .join(with_epoch, "epoch", "left")
    .orderBy("epoch")
    .withColumn("ts_resampled", col("epoch").cast("timestamp"))
    .show(15, False))

## +----------+---------------------+------+---------------------+   
## |epoch     |ts                   |val   |ts_resampled         |
## +----------+---------------------+------+---------------------+
## |1339459200|2012-06-13 00:00:00.0|0.694 |2012-06-12 02:00:00.0|
## |1339545600|null                 |null  |2012-06-13 02:00:00.0|
## |1339632000|null                 |null  |2012-06-14 02:00:00.0|
## |1339718400|null                 |null  |2012-06-15 02:00:00.0|
## |1339804800|null                 |null  |2012-06-16 02:00:00.0|
## |1339891200|null                 |null  |2012-06-17 02:00:00.0|
## |1339977600|null                 |null  |2012-06-18 02:00:00.0|
## |1340064000|2012-06-20 00:00:00.0|-2.669|2012-06-19 02:00:00.0|
## |1340150400|null                 |null  |2012-06-20 02:00:00.0|
## |1340236800|null                 |null  |2012-06-21 02:00:00.0|
## |1340323200|null                 |null  |2012-06-22 02:00:00.0|
## |1340409600|null                 |null  |2012-06-23 02:00:00.0|
## |1340496000|null                 |null  |2012-06-24 02:00:00.0|
## |1340582400|null                 |null  |2012-06-25 02:00:00.0|
## |1340668800|2012-06-27 00:00:00.0|0.245 |2012-06-26 02:00:00.0|
## +----------+---------------------+------+---------------------+

In Spark >= 3.1 replace

col("epoch").cast("timestamp")

with

from pyspark.sql.functions import timestamp_seconds

timestamp_seconds("epoch")

Using low level APIs it is possible to fill data like this as I’ve shown in my answer to Spark / Scala: forward fill with last observation. Using RDDs we could also avoid shuffling data twice (once for join, once for reordering).

But there is much more important problem here. Spark performs optimally when problem can be reduced to element wise or partition wise computations. While forward fill is the case when it is possible, as far as I am aware this is typically not the case with commonly used time series models and if some operation requires a sequential access then Spark won’t provide any benefits at all.

So if you work with series which are large enough to require distributed data structure you’ll probably want to aggregate it to some object that can be easily handled by a single machine and then use your favorite non-distributed tool to handle the rest.

If you work with multiple time series where each can be handled in memory then there is of course sparkts, but I know you’re already aware of that.

More Related Contents:

  1. Finding duplicates from large data set using Apache Spark
  2. How to melt Spark DataFrame?
  3. Using a column value as a parameter to a spark DataFrame function
  4. Find maximum row per group in Spark DataFrame
  5. Unpivot in spark-sql/pyspark
  6. Split Spark Dataframe string column into multiple columns
  7. Avoid performance impact of a single partition mode in Spark window functions
  8. How to access element of a VectorUDT column in a Spark DataFrame?
  9. Spark Dataframe validating column names for parquet writes
  10. How to check if spark dataframe is empty?
  11. How to split a list to multiple columns in Pyspark?
  12. pyspark dataframe filter or include based on list
  13. How to save/insert each DStream into a permanent table
  14. Pyspark : forward fill with last observation for a DataFrame
  15. Spark load data and add filename as dataframe column
  16. How to loop through each row of dataFrame in pyspark
  17. Fill in null with previously known good value with pyspark
  18. Why does Spark think this is a cross / Cartesian join
  19. PySpark – get row number for each row in a group
  20. How to find count of Null and Nan values for each column in a PySpark dataframe efficiently?
  21. Rename more than one column using withColumnRenamed
  22. Keep only duplicates from a DataFrame regarding some field
  23. pyspark: count distinct over a window
  24. How can I access python variable in Spark SQL?
  25. PySpark: How to fillna values in dataframe for specific columns?
  26. How to exclude multiple columns in Spark dataframe in Python
  27. Stratified sampling with pyspark
  28. How to turn off scientific notation in pyspark?
  29. spark.ml StringIndexer throws ‘Unseen label’ on fit()
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment