Skip to main content

pySpark sample code

 #!/usr/bin/env python

# coding: utf-8

# In[133]:


#https://spark.apache.org/docs/latest/api/python/getting_started/install.html to check version 

get_ipython().system('pip install pyspark')
get_ipython().system('pip install findspark')
from pyspark.sql.functions import broadcast

from pyspark.sql import SparkSession 
spark = SparkSession.builder.appName("pipeline").config("spark.hadoop.fs.file.impl", "org.apache.hadoop.fs.LocalFileSystem").getOrCreate()


# In[139]:


# ingest the sales.csv file 
df1 = spark.read.csv(r"C:\Users\keert\Downloads\sales.csv", header =True, inferSchema=True) # inferSchema is given to change everything as datatype or else everything will be become string
df1.show()
df1.printSchema()


# In[140]:


# ingest products file 
df_products = spark.read.option("multiline", "true").json(r"C:\Users\keert\Downloads\products.json") # giving option as multiline since json file as multiline 
df_products.show()
df_products.printSchema() # to see the datatypes and null values 


# In[141]:


#to ingest regions file 
df_regions = spark.read.csv(r"C:\Users\keert\Downloads\regions.csv", header =True , inferSchema = True)
df_regions.show()
df_regions.printSchema()


# In[142]:


# joining two dataframe to large dataframe to perform broadcast 
df_sale_details = df1.join(df_products, "product_id", "inner")
df_sale_details.show()


# In[143]:


# trying to optimized by partition 
optimized_df = df_sale_details.repartition(4)
optimized_df.show()


# In[129]:


# optimizing by using broadcast join 
Joined_df = optimized_df.join(broadcast(df_regions), "region_id", "inner")
Joined_df.show()


# In[144]:


# creating the temp view to show the results 

Joined_df.createOrReplaceTempView("Joined_df")

# totol sales per category
total_sales_per_category = spark.sql("SELECT category, SUM(amount) AS Total_sales FROM Joined_df GROUP BY category")
total_sales_per_category.show()
total_sales_per_category.persist() # to store the result in memory or disk for further use


# In[145]:


# Total sales per region

total_sales_per_region = spark.sql("SELECT region_id, SUM(amount) AS total_sales FROM Joined_df GROUP BY region_id")
total_sales_per_region.show()
total_sales_per_region.persist()
                                   


# In[146]:


total_sales_per_category.write.csv(r"C:\Users\keert\Downloads\final.csv", header = True , mode="overwrite") # performing the write action as result in spark 
total_sales_per_category.saveAsTextFile(r"C:\Users\keert\Downloads") # Note that PySpark requires Java 8 (except prior to 8u371), 11 or 17 with JAVA_HOME properly set. If using JDK 11, set -Dio.netty.tryReflectionSetAccessible=true for Arrow related features and refer to
Labels:

Comments

Post a Comment

Popular posts from this blog

Entity Relationship (ER) Diagram Model with DBMS Example

Reference :   Entity Relationship (ER) Diagram Model with DBMS Example What is ER Diagram? ER Diagram  stands for Entity Relationship Diagram, also known as ERD is a diagram that displays the relationship of entity sets stored in a database. In other words, ER diagrams help to explain the logical structure of databases. ER diagrams are created based on three basic concepts: entities, attributes and relationships. ER Diagrams contain different symbols that use rectangles to represent entities, ovals to define attributes and diamond shapes to represent relationships. At first look, an ER diagram looks very similar to the flowchart. However, ER Diagram includes many specialized symbols, and its meanings make this model unique. The purpose of ER Diagram is to represent the entity framework infrastructure. Entity Relationship Diagram Example Table of Content: What is ER Diagram? What is ER Model? History of ER models Why use ER Diagrams? Facts about ER Diagram Model ER Diagram...
Post a Comment
Read more

SQL Joins and advanced joins and Subqueries

  Refernce :  Expert Guide to Advanced SQL Joins: What You Need to Know It's helpful to visualize how these different SQL joins work. Here's a breakdown in a table-like format, along with explanations: SQL Join Types Overview Join Type Description Key Characteristics Use Cases INNER JOIN Returns rows where there is a match in both tables. - Shows only matching records. - Excludes unmatched rows from both tables. - Retrieving related data that exists in both tables. - Finding records with corresponding entries in another table. LEFT OUTER JOIN (LEFT JOIN) Returns all rows from the left table, and the matched rows from the right table. - Includes all records from the left table. - Fills in NULL values for columns from the right table where there's no match. - Retrieving all records from one table and their related data from another, even if some records don't have matches. - Finding records in one table that don't have corresponding entries in another. RIGHT OUTER JO...
Post a Comment
Read more

GIT BASH

  Bash Shell: Git Bash uses the Bash (Bourne Again SHell) command-line interpreter. This means you can use many of the same commands you'd find in a Linux or macOS terminal. Git Integration: Git Bash is tightly integrated with Git, making it easy to execute Git commands Essential Commands: Navigation: pwd : Prints the current working directory. ls : Lists files and directories in the current directory. cd <directory> : Changes the current directory. cd .. : Moves to the parent directory. File Management: mkdir <directory> : Creates a new directory. touch <file> : Creates a new file. rm <file> : Removes a file. rmdir <directory> : Removes an empty directory. Git Commands: git init : Initializes a new Git repository. git clone <repository URL> : Clones an existing Git repository. git status : Displays the status of your working directory. git add <file> : Adds a file to the staging area. git commit -m "commit message" : Commits chan...
Post a Comment
Read more