Accumulators details in PySpark

-

Accumulators are

  • Shared by all executors to update and add information through aggregation or computative operations
  • Only “added” through an associative and commutative operation
  • Used to perform counters (Similar to Map-reduce counters) or sum operations
  • Used with RDD and DataFrame to perform sum and counter operations

Spark by default supports creating accumulators of any numeric type and provide the capability to add custom accumulator types.

Accumulators are write-only and initialize once variables where only tasks that are running on workers are allowed to update and updates from the workers get propagated automatically to the driver program. But, only the driver program is allowed to access the Accumulator variable using the value property.

Accumulator creation

Using accumulator() from SparkContext class we can create an Accumulator in PySpark programming. Users can also create Accumulators for custom types using AccumulatorParam class of PySpark.

  • sparkContext.accumulator() is used to define accumulator variables.
  • add() function is used to add/update a value in accumulator
  • value property on the accumulator variable is used to retrieve the value from the accumulator.

We can create Accumulators in PySpark for primitive types int and float. Users can also create Accumulators for custom types using AccumulatorParam class of PySpark.

Example to create an accumulator variable “accum” of type int and using it to sum all values in an RDD

accum=sc.accumulator(0)rdd=spark.sparkContext.parallelize([1,2,3,4,5])rdd.foreach(lambda x:accum.add(x))print(accum.value) #Accessed by driver

Here, we have created an accumulator variable accum using spark.sparkContext.accumulator(0) with initial value 0. Later, we are iterating each element in an rdd using foreach() action and adding each element of rdd to accum variable. Finally, we are getting accumulator value using accum.value property.

Note that, In this example, rdd.foreach() is executed on workers and accum.value is called from PySpark driver program.

Accumulator creation by Fucntion:

accuSum=spark.sparkContext.accumulator(0)def countFun(x):    global accuSum    accuSum+=xrdd.foreach(countFun)print(accuSum.value)

We can also use accumulators to do a counters.

accumCount=spark.sparkContext.accumulator(0)rdd2=spark.sparkContext.parallelize([1,2,3,4,5])rdd2.foreach(lambda x:accumCount.add(1))print(accumCount.value)

Concrete Accumulator example:

import pysparkfrom pyspark.sql import SparkSessionspark=SparkSession.builder.appName("accumulator").getOrCreate()accum=spark.sparkContext.accumulator(0)rdd=spark.sparkContext.parallelize([1,2,3,4,5])rdd.foreach(lambda x:accum.add(x))print(accum.value)accuSum=spark.sparkContext.accumulator(0)def countFun(x):    global accuSum    accuSum+=xrdd.foreach(countFun)print(accuSum.value)accumCount=spark.sparkContext.accumulator(0)rdd2=spark.sparkContext.parallelize([1,2,3,4,5])rdd2.foreach(lambda x:accumCount.add(1))print(accumCount.value)

Refer below link for Scala Accumulator for details

In detail Accumulators explanation in Spark (Java and Scala)

 

Popular posts from this blog

How to change column name in Dataframe and selection of few columns in Dataframe using Pyspark with example

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Solution: from pyspark.sql.functions import col # this can be done without using window function creditCardData = spark.read.json(“card_transactions.json”) useridMaxSpendDF = df.groupby(‘user_id’).max(‘amount’) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“max(amount)”,”max_amount”) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“user_id”,”m_user_id”) cond = [creditCardData.user_id == useridMaxSpen...
Read more

What is Garbage collection in Spark and its impact and resolution

-
Image
Garbage Collection Spark runs on the Java Virtual Machine (JVM). Because Spark can store large amounts of data in memory, it has a major reliance on Java’s memory management and garbage collection (GC). Therefore, garbage collection  (GC) can be a major issue that can affect many Spark applications. Common symptoms of excessive GC in Spark are: Slowness of application Executor heartbeat timeout GC overhead limit exceeded error Spark’s memory-centric approach and data-intensive applications make it a more common issue than other Java applications. Thankfully, it’s easy to diagnose if your Spark application is suffering from a GC problem. The Spark UI marks executors in red if they have spent too much time doing GC. Spark executors are spending a significant amount of CPU cycles performing garbage collection. This can be determined by looking at the “Executors” tab in the Spark application UI. Spark will mark an executor in  red  if the executor has spent more than 10% of the time in gar...
Read more

Window function in PySpark with Joins example using 2 Dataframes (inner join)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Solution: from pyspark.sql.functions import col # this can be done without using window function creditCardData = spark.read.json(“card_transactions.json”) useridMaxSpendDF = df.groupby(‘user_id’).max(‘amount’) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“max(amount)”,”max_amount”) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“user_id”,”m_user_id”) cond = [creditCardData.user_id == useridMaxSpen...
Read more