Apache Spark: What is the equivalent implementation of RDD.groupByKey() using RDD.aggregateByKey()?

by

As far as I can tell there is nothing to gain* in this particular case by using aggregateByKey or a similar function. Since you’re building a list there is no “real” reduction and amount of data which has to be shuffled is more or less the same.

To really observe some performance gain you need transformations which actually reduces amount of the transfered data for example counting, computing summary statistics, finding unique elements.

Regarding differences benefits of using reduceByKey(), combineByKey(), or foldByKey() there is an important conceptual difference which is easier to see when you consider Scala API singatures.

Both reduceByKey and foldByKey map from RDD[(K, V)] to RDD[(K, V)] while the second one provides additional zero element.

reduceByKey(func: (V, V) ⇒ V): RDD[(K, V)] 
foldByKey(zeroValue: V)(func: (V, V) ⇒ V): RDD[(K, V)]

combineByKey (there is no aggregateByKey, but it is the same type of transformation) transforms from RDD[(K, V)] to RDD[(K, C)]:

combineByKey[C](
   createCombiner: (V) ⇒ C,
   mergeValue: (C, V) ⇒ C,
   mergeCombiners: (C, C) ⇒ C): RDD[(K, C)]

Going back to your example only combineByKey (and in PySpark aggregateByKey) is really applicable since you are transforming from RDD[(String, Int)] to RDD[(String, List[Int])].

While in a dynamic language like Python it is actually possible to perform such an operation using foldByKey or reduceByKey it makes semantics of the code unclear and to cite @tim-peters “There should be one– and preferably only one –obvious way to do it” [1].

Difference between aggregateByKey and combineByKey is pretty much the same as between reduceByKey and foldByKey so for a list it is mostly a matter of taste:

def merge_value(acc, x):
    acc.append(x)
    return acc

def merge_combiners(acc1, acc2):
    acc1.extend(acc2)
    return acc1

rdd = (sc.parallelize([("a", 7), ("b", 3), ("a", 8)])
   .combineByKey(
       lambda x: [x],
       lambda u, v: u + [v],
       lambda u1,u2: u1+u2))

In practice you should prefer groupByKey though. PySpark implementation is significantly more optimized compared to naive implementation like the one provided above.

1.Peters, T. PEP 20 — The Zen of Python. (2004). at https://www.python.org/dev/peps/pep-0020/

* In practice there is actually quite a lot to loose here, especially when using PySpark. Python implementation of groupByKey is significantly more optimized than naive combine by key. You can check Be Smart About groupByKey, created by me and @eliasah for an additional discussion.

More Related Contents:

  1. How do I split an RDD into two or more RDDs?
  2. Apache spark dealing with case statements
  3. Spark ALS predictAll returns empty
  4. Why is the fold action necessary in Spark?
  5. How to melt Spark DataFrame?
  6. Unpivot in spark-sql/pyspark
  7. Spark Dataframe validating column names for parquet writes
  8. How to check if spark dataframe is empty?
  9. Spark: subtract two DataFrames
  10. TypeError: Column is not iterable – How to iterate over ArrayType()?
  11. Spark performance for Scala vs Python
  12. Is groupByKey ever preferred over reduceByKey
  13. How to fix ‘TypeError: an integer is required (got type bytes)’ error when trying to run pyspark after installing spark 2.4.4
  14. Pyspark : forward fill with last observation for a DataFrame
  15. Adding a group count column to a PySpark dataframe
  16. A list as a key for PySpark’s reduceByKey
  17. Multiple Spark applications with HiveContext
  18. How to loop through each row of dataFrame in pyspark
  19. Why does sortBy transformation trigger a Spark job?
  20. PySpark DataFrames – way to enumerate without converting to Pandas?
  21. Reduce a key-value pair into a key-list pair with Apache Spark
  22. What is the difference between cache and persist?
  23. Rename more than one column using withColumnRenamed
  24. pyspark: Efficiently have partitionBy write to same number of total partitions as original table
  25. Spark gives a StackOverflowError when training using ALS
  26. AWS EMR – ModuleNotFoundError: No module named ‘pyarrow’
  27. PySpark: How to fillna values in dataframe for specific columns?
  28. Spark + s3 – error – java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found
  29. spark.ml StringIndexer throws ‘Unseen label’ on fit()
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment