Credit Card Data Analysis using PySpark (Get the total amount spent by each user for each of their cards on each category) with execution plan

-

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: Get the total amount spent by each user for each of their cards on each category

cardTnDF = spark.read.json(“card_transactions.json”)
cardTnDF.groupBy(‘user_id’,’card_num’,’category’).sum(‘amount’).collect()

Output:

[Row(user_id='U_104', card_num='C_107', category='cosmetics', sum(amount)=11818), Row(user_id='U_104', card_num='C_108', category='cosmetics', sum(amount)=9522), Row(user_id='U_103', card_num='C_105', category='children', sum(amount)=5151), Row(user_id='U_103', card_num='C_104', category='groceries', sum(amount)=8536), Row(user_id='U_101', card_num='C_102', category='groceries', sum(amount)=8074), Row(user_id='U_103', card_num='C_104', category='food', sum(amount)=10510), Row(user_id='U_101', card_num='C_101', category='children', sum(amount)=11695), Row(user_id='U_103', card_num='C_105', category='entertainment', sum(amount)=12966), Row(user_id='U_101', card_num='C_101', category='entertainment', sum(amount)=14478), Row(user_id='U_103', card_num='C_106', category='groceries', sum(amount)=6389), Row(user_id='U_103', card_num='C_105', category='food', sum(amount)=6937), Row(user_id='U_104', card_num='C_107', category='groceries', sum(amount)=12774), Row(user_id='U_104', card_num='C_107', category='entertainment', sum(amount)=7106), Row(user_id='U_101', card_num='C_101', category='groceries', sum(amount)=13529), Row(user_id='U_101', card_num='C_102', category='children', sum(amount)=14437), Row(user_id='U_101', card_num='C_101', category='food', sum(amount)=11525), Row(user_id='U_103', card_num='C_106', category='food', sum(amount)=11578), Row(user_id='U_104', card_num='C_107', category='children', sum(amount)=10151), Row(user_id='U_102', card_num='C_103', category='cosmetics', sum(amount)=25226), Row(user_id='U_101', card_num='C_102', category='food', sum(amount)=11946), Row(user_id='U_101', card_num='C_102', category='entertainment', sum(amount)=15598), Row(user_id='U_102', card_num='C_103', category='children', sum(amount)=29825), Row(user_id='U_104', card_num='C_108', category='food', sum(amount)=9905), Row(user_id='U_104', card_num='C_108', category='entertainment', sum(amount)=11241), Row(user_id='U_104', card_num='C_107', category='food', sum(amount)=14362), Row(user_id='U_103', card_num='C_105', category='cosmetics', sum(amount)=9336), Row(user_id='U_103', card_num='C_106', category='cosmetics', sum(amount)=8833), Row(user_id='U_104', card_num='C_108', category='children', sum(amount)=12042), Row(user_id='U_102', card_num='C_103', category='groceries', sum(amount)=24745), Row(user_id='U_103', card_num='C_106', category='children', sum(amount)=10154), Row(user_id='U_103', card_num='C_104', category='cosmetics', sum(amount)=8316), Row(user_id='U_104', card_num='C_108', category='groceries', sum(amount)=12018), Row(user_id='U_101', card_num='C_101', category='cosmetics', sum(amount)=15354), Row(user_id='U_103', card_num='C_104', category='entertainment', sum(amount)=5794), Row(user_id='U_102', card_num='C_103', category='food', sum(amount)=14314), Row(user_id='U_103', card_num='C_104', category='children', sum(amount)=10053), Row(user_id='U_102', card_num='C_103', category='entertainment', sum(amount)=32365), Row(user_id='U_101', card_num='C_102', category='cosmetics', sum(amount)=9148), Row(user_id='U_103', card_num='C_105', category='groceries', sum(amount)=9976), Row(user_id='U_103', card_num='C_106', category='entertainment', sum(amount)=7010)]
== Physical Plan ==AdaptiveSparkPlan (10)+- == Final Plan ==   * HashAggregate (6)   +- AQEShuffleRead (5)      +- ShuffleQueryStage (4)         +- Exchange (3)            +- * HashAggregate (2)               +- Scan json  (1)+- == Initial Plan ==   HashAggregate (9)   +- Exchange (8)      +- HashAggregate (7)         +- Scan json  (1)(1) Scan json Output [4]: [amount#480L, card_num#481, category#482, user_id#485]Batched: falseLocation: InMemoryFileIndex [file:/Users/dpq/Practice/card_transactions.json]ReadSchema: struct<amount:bigint,card_num:string,category:string,user_id:string>(2) HashAggregate [codegen id : 1]Input [4]: [amount#480L, card_num#481, category#482, user_id#485]Keys [3]: [user_id#485, card_num#481, category#482]Functions [1]: [partial_sum(amount#480L)]Aggregate Attributes [1]: [sum#504L]Results [4]: [user_id#485, card_num#481, category#482, sum#505L](3) ExchangeInput [4]: [user_id#485, card_num#481, category#482, sum#505L]Arguments: hashpartitioning(user_id#485, card_num#481, category#482, 200), ENSURE_REQUIREMENTS, [id=#643](4) ShuffleQueryStageOutput [4]: [user_id#485, card_num#481, category#482, sum#505L]Arguments: 0(5) AQEShuffleReadInput [4]: [user_id#485, card_num#481, category#482, sum#505L]Arguments: coalesced(6) HashAggregate [codegen id : 2]Input [4]: [user_id#485, card_num#481, category#482, sum#505L]Keys [3]: [user_id#485, card_num#481, category#482]Functions [1]: [sum(amount#480L)]Aggregate Attributes [1]: [sum(amount#480L)#498L]Results [4]: [user_id#485, card_num#481, category#482, sum(amount#480L)#498L AS sum(amount)#499L](7) HashAggregateInput [4]: [amount#480L, card_num#481, category#482, user_id#485]Keys [3]: [user_id#485, card_num#481, category#482]Functions [1]: [partial_sum(amount#480L)]Aggregate Attributes [1]: [sum#504L]Results [4]: [user_id#485, card_num#481, category#482, sum#505L](8) ExchangeInput [4]: [user_id#485, card_num#481, category#482, sum#505L]Arguments: hashpartitioning(user_id#485, card_num#481, category#482, 200), ENSURE_REQUIREMENTS, [id=#633](9) HashAggregateInput [4]: [user_id#485, card_num#481, category#482, sum#505L]Keys [3]: [user_id#485, card_num#481, category#482]Functions [1]: [sum(amount#480L)]Aggregate Attributes [1]: [sum(amount#480L)#498L]Results [4]: [user_id#485, card_num#481, category#482, sum(amount#480L)#498L AS sum(amount)#499L](10) AdaptiveSparkPlanOutput [4]: [user_id#485, card_num#481, category#482, sum(amount)#499L]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