How to transform data with sliding window over time series data in Pyspark

by

As far as I can tell sliding function is not available from Python and SlidingRDD is a private class and cannot be accessed outside MLlib.

If you to use sliding on an existing RDD you can create poor man’s sliding like this:

def sliding(rdd, n):
    assert n > 0
    def gen_window(xi, n):
        x, i = xi
        return [(i - offset, (i, x)) for offset in xrange(n)]

    return (
        rdd.
        zipWithIndex(). # Add index
        flatMap(lambda xi: gen_window(xi, n)). # Generate pairs with offset
        groupByKey(). # Group to create windows
        # Sort values to ensure order inside window and drop indices
        mapValues(lambda vals: [x for (i, x) in sorted(vals)]).
        sortByKey(). # Sort to makes sure we keep original order
        values(). # Get values
        filter(lambda x: len(x) == n)) # Drop beginning and end

Alternatively you can try something like this (with a small help of toolz)

from toolz.itertoolz import sliding_window, concat

def sliding2(rdd, n):
    assert n > 1

    def get_last_el(i, iter):
        """Return last n - 1 elements from the partition"""
        return  [(i, [x for x in iter][(-n + 1):])]

    def slide(i, iter):
        """Prepend previous items and return sliding window"""
        return sliding_window(n, concat([last_items.value[i - 1], iter]))

    def clean_last_items(last_items):
        """Adjust for empty or to small partitions"""
        clean = {-1: [None] * (n - 1)}
        for i in range(rdd.getNumPartitions()):
            clean[i] = (clean[i - 1] + list(last_items[i]))[(-n + 1):]
        return {k: tuple(v) for k, v in clean.items()}

    last_items = sc.broadcast(clean_last_items(
        rdd.mapPartitionsWithIndex(get_last_el).collectAsMap()))

    return rdd.mapPartitionsWithIndex(slide)

More Related Contents:

  1. How to split Vector into columns – using PySpark
  2. How to find median and quantiles using Spark
  3. Spark Dataframe distinguish columns with duplicated name
  4. creating spark data structure from multiline record
  5. How to change dataframe column names in pyspark?
  6. I can’t seem to get –py-files on Spark to work
  7. Retrieve top n in each group of a DataFrame in pyspark
  8. Pyspark: explode json in column to multiple columns
  9. How can we JOIN two Spark SQL dataframes using a SQL-esque “LIKE” criterion?
  10. Spark RDD to DataFrame python
  11. Pyspark ‘NoneType’ object has no attribute ‘_jvm’ error
  12. Why is Apache-Spark – Python so slow locally as compared to pandas?
  13. Cast column containing multiple string date formats to DateTime in Spark
  14. How to explode multiple columns of a dataframe in pyspark
  15. Spark groupByKey alternative
  16. How to extract an element from a array in pyspark
  17. PySpark create new column with mapping from a dict
  18. Tuning parameters for implicit pyspark.ml ALS matrix factorization model through pyspark.ml CrossValidator
  19. Spark iteration time increasing exponentially when using join
  20. Create Spark DataFrame. Can not infer schema for type
  21. PySpark DataFrames – way to enumerate without converting to Pandas?
  22. Getting Spark, Python, and MongoDB to work together
  23. Create single row dataframe from list of list PySpark
  24. Build a hierarchy from a relational data-set using Pyspark
  25. Concatenate two PySpark dataframes
  26. How do I convert an array (i.e. list) column to Vector
  27. Filtering DataFrame using the length of a column
  28. How to process RDDs using a Python class?
  29. What is the Spark DataFrame method `toPandas` actually doing?
  30. how to add Row id in pySpark dataframes [duplicate]

Leave a Comment