Dedup logic in Spark SQL

Dedup logic in Spark SQL or Hive:

select
    *
from (select
    *
   ,(row_number() over (partition by user_id order by mts_trckng_rowkey)) as alias_1
    from DB_NAME.TABLE_NAME
    where dt = '20191025'
) alias_2
WHERE alias_2.alias_1 = 1;

Dedup logic in Scala:

import org.apache.spark.sql.functions._

import org.apache.spark.sql.expressions._

// Dedup logic : Remove duplicate records by eventid and DataSourceKey combination

val Data_DF_Final_Dedup = Data_DF_Final.withColumn("ROWNUM", row_number().over(Window.partitionBy(col("Id"), col("DataSourceKey")).orderBy($"Updateddate".desc))).filter("ROWNUM = 1").drop("ROWNUM")

display(Data_DF_Final_Dedup)

ADB Spark SQL

%sql 

select * from

(

select  eventid, 

row_number() OVER (PARTITION BY Id ORDER BY Updateddate DESC) alias1

from parquet.`abfss://xy@abc.dfs.core.windows.net/dataproducts/test/v1/sen/full`

where id = 115894

) alias2

where alias2.alias1=1

Difference between Coalesce and Repartition

The coalesce reduces the number of partitions in a DataFrame. 

The repartition either increase or decrease the number of partitions in a DataFrame.

The repartition algorithm does a full shuffle of the data and creates equal sized partitions of data. coalesce combines existing partitions to avoid a full shuffle.

Summary Of Difference

coalesce()	repartition()
reduce the number of partitions	increase or decrease the number of partitions.
Tries to minimize data movement by avoiding network shuffle.	A network shuffle will be triggered which can increase data movement.
Creates unequal sized partitions	Creates equal sized partitions

Apache Parquet File Format

Apache Parquet is a file format. The Parquet fire format is designed as a columnar storage format to support complex data processing.

Apache Parquet is a self-describing data format which embeds the schema, or structure, within the data itself. This results in a file that is optimized for query performance and minimizing I/O. Specifically, it has the following characteristics:

• Apache Parquet is column-oriented and designed to bring efficient columnar storage of data compared to row based files like CSV
• Apache Parquet is built from the ground up with complex nested data structures in mind
• Apache Parquet is built to support very efficient compression and encoding schemes (see Google Snappy)
• Apache Parquet allows to lower storage costs for data files and maximizes the effectiveness of querying data with serverless technologies like Amazon Athena, Redshift Spectrum, BigQuery, and Azure Data Lakes.
• Licensed under the Apache software foundation and available to any project.

Adaptive Execution in Spark

Adaptive Query Execution (aka Adaptive Optimisation or Adaptive Execution) is an optimisation of a query execution plan that Spark Planner uses for allowing alternative execution plans at runtime that would be optimized better based on runtime statistics.

Quoting the description of a talk by the authors of Adaptive Query Execution:

At runtime, the adaptive execution mode can change shuffle join to broadcast join if it finds the size of one table is less than the broadcast threshold. It can also handle skewed input data for join and change the partition number of the next stage to better fit the data scale. In general, adaptive execution decreases the effort involved in tuning SQL query parameters and improves the execution performance by choosing a better execution plan and parallelism at runtime.

Adaptive Query Execution is disabled by default. Set spark.sql.adaptive.enabled configuration property to true to enable it.

References:

1. (video) An Adaptive Execution Engine For Apache Spark SQL — Carson Wang
2. An adaptive execution mode for Spark SQL by Carson Wang (Intel), Yucai Yu (Intel) at Strata Data Conference in Singapore, December 7, 2017
3. https://issues.apache.org/jira/browse/SPARK-23128
4. https://issues.apache.org/jira/browse/SPARK-9850

Broadcast Variables

Broadcast Variables

From the official documentation about Broadcast Variables:

Broadcast variables allow the programmer to keep a read-only variable cached on each machine rather than shipping a copy of it with tasks.

And later in the document:

Explicitly creating broadcast variables is only useful when tasks across multiple stages need the same data or when caching the data in deserialized form is important.

Figure 1. Broadcasting a value to executors

To use a broadcast value in a Spark transformation you have to create it first using SparkContext.broadcast and then use value method to access the shared value. Learn it in Introductory Example section.

The Broadcast feature in Spark uses SparkContext to create broadcast values and BroadcastManagerand ContextCleaner to manage their lifecycle.

Figure 2. SparkContext to broadcast using BroadcastManager and ContextCleaner

Tip	Not only can Spark developers use broadcast variables for efficient data distribution, but Spark itself uses them quite often. A very notable use case is when Spark distributes tasks to executors for their execution. That does change my perspective on the role of broadcast variables in Spark.

Broadcast Spark Developer-Facing Contract

The developer-facing Broadcast contract allows Spark developers to use it in their applications.

Table 1. Broadcast API

Method Name	Description
id	The unique identifier
value	The value
unpersist	Asynchronously deletes cached copies of this broadcast on the executors.
destroy	Destroys all data and metadata related to this broadcast variable.
toString	The string representation

Lifecycle of Broadcast Variable

You can create a broadcast variable of type T using SparkContext.broadcast method.

scala> val b = sc.broadcast(1)
b: org.apache.spark.broadcast.Broadcast[Int] = Broadcast(0)

Tip	Enable DEBUG logging level for org.apache.spark.storage.BlockManager logger to debug broadcast method. Read BlockManager to find out how to enable the logging level.

With DEBUG logging level enabled, you should see the following messages in the logs:

DEBUG BlockManager: Put block broadcast_0 locally took  430 ms
DEBUG BlockManager: Putting block broadcast_0 without replication took  431 ms
DEBUG BlockManager: Told master about block broadcast_0_piece0
DEBUG BlockManager: Put block broadcast_0_piece0 locally took  4 ms
DEBUG BlockManager: Putting block broadcast_0_piece0 without replication took  4 ms

After creating an instance of a broadcast variable, you can then reference the value using value method.

scala> b.value
res0: Int = 1

Note	value method is the only way to access the value of a broadcast variable.

With DEBUG logging level enabled, you should see the following messages in the logs:

DEBUG BlockManager: Getting local block broadcast_0
DEBUG BlockManager: Level for block broadcast_0 is StorageLevel(disk, memory, deserialized, 1 replicas)

When you are done with a broadcast variable, you should destroy it to release memory.

scala> b.destroy

With DEBUG logging level enabled, you should see the following messages in the logs:

DEBUG BlockManager: Removing broadcast 0
DEBUG BlockManager: Removing block broadcast_0_piece0
DEBUG BlockManager: Told master about block broadcast_0_piece0
DEBUG BlockManager: Removing block broadcast_0

Before destroying a broadcast variable, you may want to unpersist it.

scala> b.unpersist

Getting the Value of Broadcast Variable — value Method

value: T

value returns the value of a broadcast variable. You can only access the value until it is destroyedafter which you will see the following SparkException exception in the logs:

org.apache.spark.SparkException: Attempted to use Broadcast(0) after it was destroyed (destroy at <console>:27)
  at org.apache.spark.broadcast.Broadcast.assertValid(Broadcast.scala:144)
  at org.apache.spark.broadcast.Broadcast.value(Broadcast.scala:69)
  ... 48 elided

Internally, value makes sure that the broadcast variable is valid, i.e. destroy was not called, and, if so, calls the abstract getValue method.

Note	getValue is abstracted and broadcast variable implementations are supposed to provide a concrete behaviour. Refer to TorrentBroadcast.

Unpersisting Broadcast Variable — unpersist Methods

unpersist(): Unit
unpersist(blocking: Boolean): Unit

Destroying Broadcast Variable — destroy Method

destroy(): Unit

destroy removes a broadcast variable.

Note	Once a broadcast variable has been destroyed, it cannot be used again.

If you try to destroy a broadcast variable more than once, you will see the following SparkExceptionexception in the logs:

scala> b.destroy
org.apache.spark.SparkException: Attempted to use Broadcast(0) after it was destroyed (destroy at <console>:27)
  at org.apache.spark.broadcast.Broadcast.assertValid(Broadcast.scala:144)
  at org.apache.spark.broadcast.Broadcast.destroy(Broadcast.scala:107)
  at org.apache.spark.broadcast.Broadcast.destroy(Broadcast.scala:98)
  ... 48 elided

Internally, destroy executes the internal destroy (with blocking enabled).

Removing Persisted Data of Broadcast Variable — destroyInternal Method

destroy(blocking: Boolean): Unit

destroy destroys all data and metadata of a broadcast variable.

Note	destroy is a private[spark] method.

Internally, destroy marks a broadcast variable destroyed, i.e. the internal _isValid flag is disabled.

You should see the following INFO message in the logs:

INFO TorrentBroadcast: Destroying Broadcast([id]) (from [destroySite])

In the end, doDestroy method is executed (that broadcast implementations are supposed to provide).

Note	doDestroy is a part of the Broadcast contract for broadcast implementations so they can provide their own custom behaviour.

Introductory Example

Let’s start with an introductory example to check out how to use broadcast variables and build your initial understanding.

You’re going to use a static mapping of interesting projects with their websites, i.e. Map[String, String] that the tasks, i.e. closures (anonymous functions) in transformations, use.

scala> val pws = Map("Apache Spark" -> "http://spark.apache.org/", "Scala" -> "http://www.scala-lang.org/")
pws: scala.collection.immutable.Map[String,String] = Map(Apache Spark -> http://spark.apache.org/, Scala -> http://www.scala-lang.org/)

scala> val websites = sc.parallelize(Seq("Apache Spark", "Scala")).map(pws).collect
...
websites: Array[String] = Array(http://spark.apache.org/, http://www.scala-lang.org/)

It works, but is very ineffective as the pws map is sent over the wire to executors while it could have been there already. If there were more tasks that need the pws map, you could improve their performance by minimizing the number of bytes that are going to be sent over the network for task execution.

Enter broadcast variables.

val pwsB = sc.broadcast(pws)
val websites = sc.parallelize(Seq("Apache Spark", "Scala")).map(pwsB.value).collect
// websites: Array[String] = Array(http://spark.apache.org/, http://www.scala-lang.org/)

Semantically, the two computations - with and without the broadcast value - are exactly the same, but the broadcast-based one wins performance-wise when there are more executors spawned to execute many tasks that use pws map.

Introduction

Broadcast is part of Spark that is responsible for broadcasting information across nodes in a cluster.

You use broadcast variable to implement map-side join, i.e. a join using a map. For this, lookup tables are distributed across nodes in a cluster using broadcast and then looked up inside map (to do the join implicitly).

When you broadcast a value, it is copied to executors only once (while it is copied multiple times for tasks otherwise). It means that broadcast can help to get your Spark application faster if you have a large value to use in tasks or there are more tasks than executors.

It appears that a Spark idiom emerges that uses broadcast with collectAsMap to create a Map for broadcast. When an RDD is map over to a smaller dataset (column-wise not record-wise), collectAsMap, and broadcast, using the very big RDD to map its elements to the broadcast RDDs is computationally faster.

val acMap = sc.broadcast(myRDD.map { case (a,b,c,b) => (a, c) }.collectAsMap)
val otherMap = sc.broadcast(myOtherRDD.collectAsMap)

myBigRDD.map { case (a, b, c, d) =>
  (acMap.value.get(a).get, otherMap.value.get(c).get)
}.collect

Use large broadcasted HashMaps over RDDs whenever possible and leave RDDs with a key to lookup necessary data as demonstrated above.

Spark comes with a BitTorrent implementation.

It is not enabled by default.

Broadcast Contract

The Broadcast contract is made up of the following methods that custom Broadcastimplementations are supposed to provide:

1. getValue
2. doUnpersist
3. doDestroy

Note	TorrentBroadcast is the only implementation of the Broadcast contract.

Note	Broadcast Spark Developer-Facing Contract is the developer-facing Broadcast contract that allows Spark developers to use it in their applications.