Python Spark Cumulative Sum by Group Using DataFrame

by

This can be done using a combination of a window function and the Window.unboundedPreceding value in the window’s range as follows:

from pyspark.sql import Window
from pyspark.sql import functions as F

windowval = (Window.partitionBy('class').orderBy('time')
             .rangeBetween(Window.unboundedPreceding, 0))
df_w_cumsum = df.withColumn('cum_sum', F.sum('value').over(windowval))
df_w_cumsum.show()

+----+-----+-----+-------+
|time|value|class|cum_sum|
+----+-----+-----+-------+
|   1|    3|    b|      3|
|   2|    3|    b|      6|
|   1|    2|    a|      2|
|   2|    2|    a|      4|
|   3|    2|    a|      6|
+----+-----+-----+-------+

More Related Contents:

  1. Finding duplicates from large data set using Apache Spark
  2. How to improve performance for slow Spark jobs using DataFrame and JDBC connection?
  3. TypeError: Column is not iterable – How to iterate over ArrayType()?
  4. How to save/insert each DStream into a permanent table
  5. Pyspark: Pass multiple columns in UDF
  6. Apache spark dealing with case statements
  7. Find maximum row per group in Spark DataFrame
  8. How do I split an RDD into two or more RDDs?
  9. Spark Dataframe validating column names for parquet writes
  10. Spark RDD to DataFrame python
  11. How to split a list to multiple columns in Pyspark?
  12. Link Spark with iPython Notebook
  13. Spark parquet partitioning : Large number of files
  14. Spark load data and add filename as dataframe column
  15. PySpark: how to resample frequencies
  16. Efficient pyspark join
  17. Explode array data into rows in spark [duplicate]
  18. Why does Spark think this is a cross / Cartesian join
  19. Apache Spark: What is the equivalent implementation of RDD.groupByKey() using RDD.aggregateByKey()?
  20. How to find count of Null and Nan values for each column in a PySpark dataframe efficiently?
  21. pyspark: count distinct over a window
  22. pyspark: Efficiently have partitionBy write to same number of total partitions as original table
  23. Serialize a custom transformer using python to be used within a Pyspark ML pipeline
  24. AWS EMR – ModuleNotFoundError: No module named ‘pyarrow’
  25. Not able to cat dbfs file in databricks community edition cluster. FileNotFoundError: [Errno 2] No such file or directory:
  26. How to exclude multiple columns in Spark dataframe in Python
  27. Spark + s3 – error – java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found
  28. Stratified sampling with pyspark
  29. Get CSV to Spark dataframe
  30. spark.ml StringIndexer throws ‘Unseen label’ on fit()

Leave a Comment