[Day 177] Spark for batch processing

 Hello :)
Today is Day 177!


A quick summary of today:


What is batch processing of data?

  • method of executing data processing tasks on a large volume of collected data all at once, rather than in real-time. It is often done at scheduled times (i.e. hourly, daily, weekly, x times per hour or minutes) or when sufficient data is accumulated
Technologies used for batch processing:
  • python scripts
  • SQL
  • Spark
  • flink
Advantages of batch jobs:
  • easy to manage
  • retry
  • scale
Disadvantages:
  • delay in getting fresh data
Spark is the most popular batch processing tool, and its variation in PySpark is popular. I have used PySpark during my placement at Lloyds Banking Group back in 2019-2020 so there was not much new info~
The most important bit is that spark works with clusters and in order to utilise spark's power in handling large datasets, we need to partition our data. 

Then, I got a refresher on spark cluster (image is from google)

We are the guest where we connect to a SparkContext and execute functions. In the spark cluster there is a cluster master that manages the workload among workers. In turn, each worker node runs multiple executors (individual tasks) and store partitions of data.

Spark groupBy (thank you GPT for the nice summary)
Stage Creation:
  • Spark divides the computation into stages based on the shuffle boundaries.
  • For a group by operation, data often needs to be shuffled so that records with the same key end up on the same partition
Shuffling:
  • Spark performs a shuffle operation to redistribute data across the cluster.
  • Each partition will send its records to the appropriate partition based on the hash of the group by key.
  • This step is expensive because it involves disk I/O and network I/O
Aggregation:
  • After the shuffle, each node performs a local aggregation on its partition of the data.
  • Spark uses combiners to perform partial aggregation before shuffling to reduce the amount of data transferred
Result Collection:
  • The results from each node are combined to form the final output

Logical Plan:

Spark join

Spark creates a logical plan based on the join operation specified. This plan is then optimized.

Physical Plan:

  • The logical plan is converted into a physical plan that specifies how the join should be executed (e.g., broadcast join, sort-merge join, etc.)

Execution Strategy:

  • Broadcast Join: If one of the DataFrames is small enough, it can be broadcasted to all worker nodes to avoid shuffling
  • Sort-Merge Join: Both DataFrames are sorted by the join key and then merged. This requires shuffling both DataFrames
  • Shuffle Hash Join: DataFrames are shuffled based on the join key, and a hash table is built for the join

Shuffling:

  • Depending on the join type, Spark may shuffle the data across nodes to ensure that matching keys from both DataFrames are on the same partition

Joining:

  • The actual join operation is performed on the partitions, and results are combined


Then using a download bash script, I downloaded some taxi data as .csv.gz and then using spark, uploaded it to GCS. 

Bash script to download the data (ran it for green and yellow taxi for both 2020 and 2021):

Code to transform to parquet

I installed the gcloud cli and uploaded the data through it ~ here is the data in GCS (in the pq folder)
Next, to load data from gcs using pyspark, first I needed to install an extra google cloud util for hadoop so that it can read gs://mage-zoomcamp-ii/.... And after some debugging on spark not being able to connect to an IP ~

Creating a context
And connecting and getting the data ~
The spark debugging took me a lot of time ~ it had to do with the default ip (127.0.0.1) I had set in spark-defaults.conf for spark to connect to, and because I was trying to use GCP, it could not connect so I removed the IP, and now it is just (pic) and it works. 

This is not all from this module, there is a bit left, but I will leave that for tomorrow. I uploaded the code I used today to my repo. 


Also, a small update from the lab ~ I wrote the first version of my (at the moment titled) 'Previous works' section. ^^


That is all for today!

See you tomorrow :) 

Popular posts from this blog

[Day 198] Transactions Data Streaming Pipeline Porject [v1 completed]

Image
 Hello :) Today is Day 198! A quick summary of today: data streaming pipeline project [v1 done] Here is a link to the project's repo. Well ... I did not know I can do it in a day (~14 hours) after yesterday's issues but here we are. Turns out in order to insert the full (~70 variables with nested/list structure), I need the proper pyspark schema. And yesterday I did not have that and that is why when I was reading data in the kafka producer I was getting NULL in the columns - my schema was wrong. Well today I not only fixed the schema for the 4 variables I had yesterday, but included *all* the variables that come from the Stripe API ~ 70 (for completeness).  When I run docker-compose, the data streams and is input into the postgres db (and is still running). Unfortunately, the free Stripe API for creating realistic transactions has a limit of 25, so every 3 seconds, 25 new transactions are sent to the db. It has been running half the day (since I got that set up) and as I am wr
Read more

[Day 107] Transforming natural language to charts

Image
 Hello :) Today is Day 107! A quick summary of today: created text2chart - transforming natural language to charts [ github ] [ webapp ] I am super sick, but in the around 3 hours of my fever not being high, I managed to study a bit. I was again thinking about creating some nlp project but was stuck on the idea of how to host a model. After looking around I found this article where they create a similar app called chat2vis  and they use codellama. I know codellama is an open source model that I can get from huggingface and I wondered, how did they host it. Turns out - they ask the user to input a huggingface api key, which then allows for querying and generating content. (AWESOME! this is applicable to some of the other PDF chat apps that I have, but I will look to make an app of them later).  My fever eased a bit, and I sat on my chair to do some work ^^ Here is a breakdown of the app. I used streamlit. Also, 2 default datasets during dev: Financial Statements of Major Companies(2009-
Read more

[Day 54] I became a backprop ninja! (woohoo)

Image
 Hello :) Today is Day 54! A quick summary of today: Read through cs231n's neural network lectures [1] [2] [3] Spent 8 hours going over Andrej Karpathy's backprop ninja tutorial After yesterday's confidence boost in doing backprop, I came in today with the intention of becoming of, as Andrej Karpathy calls it, a backprop ninja.  The cs231n NN lectures were the usual about how NN became to be, param initialization, activation functions, loss functions, regularization and optimizers. Nothing out of the ordinary, but still was good to reassure myself of that knowledge. But that was definitely not the main even of today. As I mentioned, I wanted to become (or at least attempt to become) a backprop ninja today. I have watched that particular tutorial twice, the 1st time I had no idea what is going on, the 2nd time I could follow, and now it was the 3rd time - and now, I wanted to participate and actually see the why and the how of the things we have to do, when we do manual back
Read more