Reduce a key-value pair into a key-list pair with Apache Spark

by

Map and ReduceByKey

Input type and output type of reduce must be the same, therefore if you want to aggregate a list, you have to map the input to lists. Afterwards you combine the lists into one list.

Combining lists

You’ll need a method to combine lists into one list. Python provides some methods to combine lists.

append modifies the first list and will always return None.

x = [1, 2, 3]
x.append([4, 5])
# x is [1, 2, 3, [4, 5]]

extend does the same, but unwraps lists:

x = [1, 2, 3]
x.extend([4, 5])
# x is [1, 2, 3, 4, 5]

Both methods return None, but you’ll need a method that returns the combined list, therefore just use the plus sign.

x = [1, 2, 3] + [4, 5]
# x is [1, 2, 3, 4, 5]

Spark

file = spark.textFile("hdfs://...")
counts = file.flatMap(lambda line: line.split(" ")) \
         .map(lambda actor: (actor.split(",")[0], actor)) \ 

         # transform each value into a list
         .map(lambda nameTuple: (nameTuple[0], [ nameTuple[1] ])) \

         # combine lists: ([1,2,3] + [4,5]) becomes [1,2,3,4,5]
         .reduceByKey(lambda a, b: a + b)

CombineByKey

It’s also possible to solve this with combineByKey, which is used internally to implement reduceByKey, but it’s more complex and “using one of the specialized per-key combiners in Spark can be much faster”. Your use case is simple enough for the upper solution.

GroupByKey

It’s also possible to solve this with groupByKey, but it reduces parallelization and therefore could be much slower for big data sets.

More Related Contents:

  1. How to find median and quantiles using Spark
  2. ‘PipelinedRDD’ object has no attribute ‘toDF’ in PySpark
  3. Spark union of multiple RDDs
  4. Spark groupByKey alternative
  5. How to extract an element from a array in pyspark
  6. A list as a key for PySpark’s reduceByKey
  7. PySpark DataFrames – way to enumerate without converting to Pandas?
  8. Random numbers generation in PySpark
  9. How to split Vector into columns – using PySpark
  10. creating spark data structure from multiline record
  11. I can’t seem to get –py-files on Spark to work
  12. Retrieve top n in each group of a DataFrame in pyspark
  13. Pyspark: explode json in column to multiple columns
  14. Spark: Broadcast variables: It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transforamtion
  15. Pyspark ‘NoneType’ object has no attribute ‘_jvm’ error
  16. Cast column containing multiple string date formats to DateTime in Spark
  17. How to explode multiple columns of a dataframe in pyspark
  18. PySpark: multiple conditions in when clause
  19. Create Spark DataFrame. Can not infer schema for type
  20. Load CSV file with PySpark
  21. What is the best way to remove accents with Apache Spark dataframes in PySpark?
  22. ImportError: No module named numpy on spark workers
  23. How to pass a constant value to Python UDF?
  24. How to zip two array columns in Spark SQL
  25. Build a hierarchy from a relational data-set using Pyspark
  26. Concatenate two PySpark dataframes
  27. How to process RDDs using a Python class?
  28. How to pivot on multiple columns in Spark SQL?
  29. What is the Spark DataFrame method `toPandas` actually doing?
  30. how to add Row id in pySpark dataframes [duplicate]

Leave a Comment