Credit Card Data Analysis using PySpark (how to use auto broadcast join after disabling it)

-

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|

Application: Here we will disable auto broadcast and then join 2 data frames and will see execution plan which should use broadcast join

Solution:

from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast
spark = SparkSession.builder.master(‘local[2]’)\
.appName(‘RDD_Methods_Examples’)\
.getOrCreate()

print(spark.conf.get(“spark.sql.autoBroadcastJoinThreshold”))
spark.conf.set(“spark.sql.autoBroadcastJoinThreshold”, -1)
print(spark.conf.get(“spark.sql.autoBroadcastJoinThreshold”))
creditCardData = spark.read.json(“card_transactions.json”)
userIdToAmountMapping = {}

useridMaxSpendDF = creditCardData.groupby(‘user_id’).max(‘amount’)

useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“max(amount)”,”max_amount”)
useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“user_id”,”m_user_id”)
cond = [creditCardData.user_id == useridMaxSpendDF.m_user_id, creditCardData.amount == useridMaxSpendDF.max_amount]
joinedData = creditCardData.join(broadcast(useridMaxSpendDF),cond,”inner”)
joinedData.show()

-1-1+------+--------+-------------+--------+----------+-------+---------+----------+|amount|card_num|     category|merchant|        ts|user_id|m_user_id|max_amount|+------+--------+-------------+--------+----------+-------+---------+----------+|  1000|   C_101|entertainment|   M_100|1580163399|  U_101|    U_101|      1000||   997|   C_103|    groceries|   M_103|1582876481|  U_102|    U_102|       997||   977|   C_104|    groceries|   M_101|1579402924|  U_103|    U_103|       977||   977|   C_105|         food|   M_102|1581369586|  U_103|    U_103|       977||   977|   C_106|         food|   M_100|1580897199|  U_103|    U_103|       977||   996|   C_108|         food|   M_106|1581391534|  U_104|    U_104|       996||   996|   C_107|     children|   M_107|1580776821|  U_104|    U_104|       996|+------+--------+-------------+--------+----------+-------+---------+----------+
== Physical Plan ==AdaptiveSparkPlan (26)+- == Final Plan ==   CollectLimit (15)   +- * Project (14)      +- * BroadcastHashJoin Inner BuildRight (13)         :- * Filter (2)         :  +- Scan json  (1)         +- BroadcastQueryStage (12)            +- BroadcastExchange (11)               +- * Filter (10)                  +- * HashAggregate (9)                     +- AQEShuffleRead (8)                        +- ShuffleQueryStage (7)                           +- Exchange (6)                              +- * HashAggregate (5)                                 +- * Filter (4)                                    +- Scan json  (3)+- == Initial Plan ==   CollectLimit (25)   +- Project (24)      +- BroadcastHashJoin Inner BuildRight (23)         :- Filter (16)         :  +- Scan json  (1)         +- BroadcastExchange (22)            +- Filter (21)               +- HashAggregate (20)                  +- Exchange (19)                     +- HashAggregate (18)                        +- Filter (17)                           +- Scan json  (3)(1) Scan json Output [6]: [amount#1294L, card_num#1295, category#1296, merchant#1297, ts#1298L, user_id#1299]Batched: falseLocation: InMemoryFileIndex [file:/Users/dpq/Practice/card_transactions.json]PushedFilters: [IsNotNull(user_id), IsNotNull(amount)]ReadSchema: struct<amount:bigint,card_num:string,category:string,merchant:string,ts:bigint,user_id:string>(2) Filter [codegen id : 3]Input [6]: [amount#1294L, card_num#1295, category#1296, merchant#1297, ts#1298L, user_id#1299]Condition : (isnotnull(user_id#1299) AND isnotnull(amount#1294L))(3) Scan json Output [2]: [amount#1322L, user_id#1327]Batched: falseLocation: InMemoryFileIndex [file:/Users/dpq/Practice/card_transactions.json]PushedFilters: [IsNotNull(user_id)]ReadSchema: struct<amount:bigint,user_id:string>(4) Filter [codegen id : 1]Input [2]: [amount#1322L, user_id#1327]Condition : isnotnull(user_id#1327)(5) HashAggregate [codegen id : 1]Input [2]: [amount#1322L, user_id#1327]Keys [1]: [user_id#1327]Functions [1]: [partial_max(amount#1322L)]Aggregate Attributes [1]: [max#1368L]Results [2]: [user_id#1327, max#1369L](6) ExchangeInput [2]: [user_id#1327, max#1369L]Arguments: hashpartitioning(user_id#1327, 200), ENSURE_REQUIREMENTS, [id=#2250](7) ShuffleQueryStageOutput [2]: [user_id#1327, max#1369L]Arguments: 0(8) AQEShuffleReadInput [2]: [user_id#1327, max#1369L]Arguments: coalesced(9) HashAggregate [codegen id : 2]Input [2]: [user_id#1327, max#1369L]Keys [1]: [user_id#1327]Functions [1]: [max(amount#1322L)]Aggregate Attributes [1]: [max(amount#1322L)#1312L]Results [2]: [user_id#1327 AS m_user_id#1319, max(amount#1322L)#1312L AS max_amount#1316L](10) Filter [codegen id : 2]Input [2]: [m_user_id#1319, max_amount#1316L]Condition : isnotnull(max_amount#1316L)(11) BroadcastExchangeInput [2]: [m_user_id#1319, max_amount#1316L]Arguments: HashedRelationBroadcastMode(List(input[0, string, true], input[1, bigint, false]),false), [id=#2303](12) BroadcastQueryStageOutput [2]: [m_user_id#1319, max_amount#1316L]Arguments: 1(13) BroadcastHashJoin [codegen id : 3]Left keys [2]: [user_id#1299, amount#1294L]Right keys [2]: [m_user_id#1319, max_amount#1316L]Join condition: None(14) Project [codegen id : 3]Output [8]: [cast(amount#1294L as string) AS amount#1352, card_num#1295, category#1296, merchant#1297, cast(ts#1298L as string) AS ts#1356, user_id#1299, m_user_id#1319, cast(max_amount#1316L as string) AS max_amount#1359]Input [8]: [amount#1294L, card_num#1295, category#1296, merchant#1297, ts#1298L, user_id#1299, m_user_id#1319, max_amount#1316L](15) CollectLimitInput [8]: [amount#1352, card_num#1295, category#1296, merchant#1297, ts#1356, user_id#1299, m_user_id#1319, max_amount#1359]Arguments: 21(16) FilterInput [6]: [amount#1294L, card_num#1295, category#1296, merchant#1297, ts#1298L, user_id#1299]Condition : (isnotnull(user_id#1299) AND isnotnull(amount#1294L))(17) FilterInput [2]: [amount#1322L, user_id#1327]Condition : isnotnull(user_id#1327)(18) HashAggregateInput [2]: [amount#1322L, user_id#1327]Keys [1]: [user_id#1327]Functions [1]: [partial_max(amount#1322L)]Aggregate Attributes [1]: [max#1368L]Results [2]: [user_id#1327, max#1369L](19) ExchangeInput [2]: [user_id#1327, max#1369L]Arguments: hashpartitioning(user_id#1327, 200), ENSURE_REQUIREMENTS, [id=#2226](20) HashAggregateInput [2]: [user_id#1327, max#1369L]Keys [1]: [user_id#1327]Functions [1]: [max(amount#1322L)]Aggregate Attributes [1]: [max(amount#1322L)#1312L]Results [2]: [user_id#1327 AS m_user_id#1319, max(amount#1322L)#1312L AS max_amount#1316L](21) FilterInput [2]: [m_user_id#1319, max_amount#1316L]Condition : isnotnull(max_amount#1316L)(22) BroadcastExchangeInput [2]: [m_user_id#1319, max_amount#1316L]Arguments: HashedRelationBroadcastMode(List(input[0, string, true], input[1, bigint, false]),false), [id=#2230](23) BroadcastHashJoinLeft keys [2]: [user_id#1299, amount#1294L]Right keys [2]: [m_user_id#1319, max_amount#1316L]Join condition: None(24) ProjectOutput [8]: [cast(amount#1294L as string) AS amount#1352, card_num#1295, category#1296, merchant#1297, cast(ts#1298L as string) AS ts#1356, user_id#1299, m_user_id#1319, cast(max_amount#1316L as string) AS max_amount#1359]Input [8]: [amount#1294L, card_num#1295, category#1296, merchant#1297, ts#1298L, user_id#1299, m_user_id#1319, max_amount#1316L](25) CollectLimitInput [8]: [amount#1352, card_num#1295, category#1296, merchant#1297, ts#1356, user_id#1299, m_user_id#1319, max_amount#1359]Arguments: 21(26) AdaptiveSparkPlanOutput [8]: [amount#1352, card_num#1295, category#1296, merchant#1297, ts#1356, user_id#1299, m_user_id#1319, max_amount#1359]Arguments: isFinalPlan=true

Popular posts from this blog

In detail Accumulators explanation in Spark (Java and Scala)

-
Accumulators are Shared variables in Spark Only “added” through an associative and commutative operation Used to perform counters (Similar to Map-reduce counters) or sum operations Spark by default supports creating accumulators of any numeric type and provide the capability to add custom accumulator types. Following accumulators can be created by Developers named accumulators – can be seen on Spark web UI under the “Accumulator” tab, On this tab, you will see two tables; the first table “accumulable” – consists of all named accumulator variables and their values. And on the second table “Tasks” – value for each accumulator modified by a task unnamed accumulators – there are not shown to UI so for all practical purposes it is advisable to use name accumulators Spark by default provides accumulator methods for long, double and collection types For example, you can create long accumulator on spark-shell using scala > val accum = sc . longAccumulator ( "SumAccumulator" ) a
Read more

Hive joins in details with examples

-
Hive Join | HiveQL Select Joins Query | Types of Join in Hive In  Apache Hive , for combining specific fields from two tables by using values common to each one we use Hive Join – HiveQL Select Joins Query. However, we need to know the syntax of Hive Join for implementation purpose. So, in this article, “Hive Join – HiveQL Select Joins Query and its types” we will cover syntax of joins in hive. Also, we will learn an example of Hive Join to understand well. Moreover, there are several types of Hive join – HiveQL Select Joins: Hive inner join, hive left outer join, hive right outer join, and hive full outer join. We will also learn Hive Join tables in depth. Apache Hive Join – HiveQL Select Joins Query Basically, for combining specific fields from two tables by using values common to each one we use Hive JOIN clause. n other words, to combine records from two or more tables in the database we use JOIN clause. However, it is more or less similar to SQL JOIN. Also, we use it to combine ro
Read more

RDD joins using PySpark examples

-
Full Outer Join: Final result will return all records from both RDDs   rdd1 = sc.parallelize([(101,’John’),(102,’Matthew’),(103,’Aakash’),(104,’Sandeep’) ,(105,’Sakura’),(106,’Abed’),(107,’Mary’),(108,’Kamla’)],2) rdd2 = sc.parallelize([(101,’USA’),(108,’UK’),(105,’Japan’),(102,’Sri Lanka’) ,(103,’India’),(104,’UAE’),(107,’Germany’),(110,’Australia’)],2) #print(rdd1.glom().collect()) #print(rdd2.glom().collect()) fullOuterJoin = rdd1.fullOuterJoin(rdd2) print(fullOuterJoin.collect()) Output: [(104, ('Sandeep', 'UAE')), (108, ('Kamla', 'UK')), (101, ('John', 'USA')), (105, ('Sakura', 'Japan')), (102, ('Matthew', 'Sri Lanka')), (106, ('Abed', None)), (110, (None, 'Australia')), (103, ('Aakash', 'India')), (107, ('Mary', 'Germany'))]   Inner Join: Final result will return matching records from both RDDs rdd1 = sc.parallelize([(101,’John’),(102,’Matthew’),(103,’Aa
Read more