Built with Tutify.io

    AIStudyLabs

    Spark RDD reduce() function example

    Spark RDD reduce() function example

    Spark RDD reduce() aggregate action function is used to calculate min, max, and total of elements in a dataset, In this tutorial, I will explain RDD reduce function syntax and usage with scala language and the same approach could be used with Java and PySpark (python) languages.

    <a id="h-syntax"></a>

    Syntax

    
def reduce(f: (T, T) => T): T

    <a id="h-usage"></a>

    Usage

    RDD reduce() function takes function type as an argument and returns the RDD with the same type as input. It reduces the elements of the input RDD using the binary operator specified.

    <a id="h-reduce-a-list-calculate-min-max-and-total-of-elements"></a>

    Reduce a list - Calculate min, max, and total of elements

    Below snippet reduces the collection for sum, minimum and maximum

    
  val listRdd = spark.sparkContext.parallelize(List(1,2,3,4,5,3,2))
  println("output min using binary : "+listRdd.reduce(_ min _))
  println("output max using binary : "+listRdd.reduce(_ max _))
  println("output sum using binary : "+listRdd.reduce(_ + _))

    Yields below output

    
output min using binary : 1
output max using binary : 5
output sum using binary : 20

    Alternatively, you can also write the above operations as below.

    
  val listRdd = spark.sparkContext.parallelize(List(1,2,3,4,5,3,2))
  println("output min : "+listRdd.reduce( (a,b) => a min b))
  println("output max : "+listRdd.reduce( (a,b) => a max b))
  println("output sum : "+listRdd.reduce( (a,b) => a + b))

    Yields below output

    
output min : 1
output max : 5
output sum : 20

    <a id="h-reduce-function-on-tupple-rdd-string-int"></a>

    Reduce function on Tupple RDD(String,Int)

    In this example, you will learn how to use reduce function on a complex RDD, which mostly used in production applications.

    
  val inputRDD = spark.sparkContext.parallelize(List(("Z", 1),("A", 20),("B", 30),("C", 40),("B", 30),("B", 60)))

  println("output min : "+inputRDD.reduce( (a,b)=> ("max",a._2 min b._2))._2)
  println("output max : "+inputRDD.reduce( (a,b)=> ("max",a._2 max b._2))._2)
  println("output sum : "+inputRDD.reduce( (a,b)=> ("Sum",a._2 + b._2))._2)

    Yields below output

    
output min : 1
output max : 60
output sum : 181

    This complete example is available at GitHub project for reference.

    <a id="h-points-to-note"></a>

    Points to Note

    <a id="h-complete-example"></a>

    Complete example

    
package com.aistudylabs.spark.rdd.functions

import org.apache.spark.sql.SparkSession

object reduceExample extends App {

  val spark = SparkSession.builder()
    .appName("AIStudyLabs.com")
    .master("local[3]")
    .getOrCreate()

  spark.sparkContext.setLogLevel("ERROR")

  val listRdd = spark.sparkContext.parallelize(List(1,2,3,4,5,3,2))

  println("output sum using binary : "+listRdd.reduce(_ min _))
  println("output min using binary : "+listRdd.reduce(_ max _))
  println("output max using binary : "+listRdd.reduce(_ + _))

  // Alternatively you can write
  println("output min : "+listRdd.reduce( (a,b) => a min b))
  println("output max : "+listRdd.reduce( (a,b) => a max b))
  println("output sum : "+listRdd.reduce( (a,b) => a + b))

  val inputRDD = spark.sparkContext.parallelize(List(("Z", 1),("A", 20),("B", 30),
    ("C", 40),("B", 30),("B", 60)))

  println("output max : "+inputRDD.reduce( (a,b)=> ("max",a._2 min b._2))._2)
  println("output max : "+inputRDD.reduce( (a,b)=> ("max",a._2 max b._2))._2)
  println("output sum : "+inputRDD.reduce( (a,b)=> ("Sum",a._2 + b._2))._2)
}

    Happy learning !!

    <a id="h-related-articles"></a>

    Related Articles