Course 1: PySpark Foundations

Week 1: Introduction to PySpark & Apache Spark

Introduction: PySpark is the Python API for Apache Spark, a powerful distributed computing framework designed for big data processing. This week introduces Apache Spark, its core concepts, and how PySpark enables Python developers to leverage Spark’s capabilities. The focus is on understanding Spark’s role in big data and running a simple PySpark program to process a dataset.

Learning Objectives: By the end of this week, you will be able to:

Understand Apache Spark’s purpose and key features.
Explain PySpark’s role as a Python interface to Spark.
Identify use cases for PySpark in big data processing.
Write and execute a basic PySpark program.
Validate PySpark setup and output.

Scope: This week introduces Apache Spark and PySpark, setting the stage for Week 2’s environment setup, Week 3’s RDDs, and subsequent topics like DataFrames, cluster architecture, and data manipulation. It emphasizes practical usage through a simple PySpark script.

Background Information: Apache Spark and PySpark are foundational for big data processing:

Apache Spark: An open-source, distributed computing framework for large-scale data processing. Key features: Speed: In-memory processing, up to 100x faster than Hadoop MapReduce. Scalability: Handles petabytes of data across clusters. Ease of Use: APIs in Python (PySpark), Scala, Java, R. Unified Engine: Supports batch, streaming, SQL, and machine learning. Core components: Spark Core, Spark SQL, Spark Streaming, MLlib, GraphX.
PySpark: Python API for Spark, enabling Python developers to write Spark applications. Integrates with Python libraries (e.g., pandas, NumPy) and Jupyter notebooks. Uses Spark’s distributed computing while leveraging Python’s simplicity.
Use Cases: Processing large datasets (e.g., sales logs, sensor data). Real-time analytics (e.g., fraud detection). ETL (Extract, Transform, Load) pipelines for data warehousing.
Challenges: Requires understanding distributed computing concepts. Setup complexity (addressed in Week 2). Memory management for large datasets.

Hands-On Example:

Scenario: You’re a data analyst tasked with using PySpark to analyze a small sales dataset. You’ll write a PySpark script to count the number of sales records and calculate total sales, demonstrating Spark’s distributed processing capabilities.

Prerequisites:

Python 3.8+ installed.
PySpark installed (basic setup for this week; detailed in Week 2): Install via pip: pip install pyspark.
Jupyter Notebook (optional, for interactive coding): pip install jupyter.
Sample dataset (sales_data.csv):sale_id,product,price,region 1,Laptop,2000,North 2,Mouse,600,North 3,Keyboard,750,South 4,Tablet,1200,West

Step-by-Step Instructions:

Set Up the Environment:

Create a working directory (e.g., pyspark_course).

Save sales_data.csv in the directory.

Open a terminal, verify Python and PySpark:python --version pip show pyspark

Launch Jupyter Notebook (or use a Python script):jupyter notebook

Create a new notebook named “Week1_PySpark_Intro”.

Write a Basic PySpark Program:

In the notebook, create a cell with the following code to initialize PySpark and process the sales data:from pyspark.sql import SparkSession # Initialize SparkSession spark = SparkSession.builder \ .appName("SalesDataAnalysis") \ .getOrCreate() # Read CSV file df = spark.read.csv("sales_data.csv", header=True, inferSchema=True) # Show the data df.show() # Count total records record_count = df.count() print(f"Total sales records: {record_count}") # Calculate total sales total_sales = df.select("price").agg({"price": "sum"}).collect()[0][0] print(f"Total sales amount: ${total_sales}") # Stop SparkSession spark.stop()

Explanation:

SparkSession: Entry point for PySpark, manages Spark application.

read.csv: Loads CSV with headers and infers data types (e.g., price as integer).

show(): Displays the DataFrame (first 20 rows).

count(): Counts rows in a distributed manner.

agg: Aggregates price column to compute sum.

stop(): Closes SparkSession to free resources.

Execute the Program:

Run the cell in Jupyter (Shift+Enter).

Check for errors (e.g., file not found, PySpark not installed).

Validate Results:

Confirm DataFrame display matches CSV content.

Verify record count (4) and total sales ($2,000 + $600 + $750 + $1,200 = $4,550).

Check Spark UI (if enabled):

Run spark.sparkContext.uiWebUrl to get URL (e.g., http://localhost:4040).

Open in browser, verify job execution (1 job for show, 1 for count, 1 for agg).

If issues occur:

Ensure sales_data.csv is in the correct directory.

Check PySpark installation (import pyspark should not fail).

Verify file path (use absolute path if needed, e.g., /path/to/sales_data.csv).

Experiment with PySpark:

Add a filter to count North region sales:north_count = df.filter(df.region == "North").count() print(f"North region sales: {north_count}")

Run, verify: North region sales: 2.

Save output to a new CSV:df.write.csv("output_sales", header=True, mode="overwrite") spark.stop()

Check output_sales directory for part files (e.g., part-00000-*.csv).

Open, verify data matches input CSV.

Clean Up:

Stop the SparkSession (already in code).

Close Jupyter Notebook.

Delete output_sales directory if not needed (rm -r output_sales).

Output: A working PySpark program that reads, processes, and analyzes a sales dataset. Proficiency in initializing PySpark, loading data, and performing basic operations. Validated output: 4 records, $4,550 total sales, 2 North sales. Saved output CSV and familiarity with Spark’s distributed processing.

PySpark Tips:

Always initialize SparkSession at the start and stop it at the end.
Use show() for quick data inspection during development.
Check Spark UI for job performance insights.
Save outputs to temporary directories for testing.
Refer to PySpark docs: https://spark.apache.org/docs/latest/api/python/

Interpretation: This hands-on example demonstrates how to use PySpark to process a small dataset, introducing Spark’s distributed computing capabilities. By validating the program’s output, you gain practical experience with PySpark’s core functionality, preparing for Week 2’s environment setup.

Supplemental Information: Apache Spark Overview: https://spark.apache.org/docs/latest/. PySpark Documentation: https://spark.apache.org/docs/latest/api/python/. Getting Started: https://spark.apache.org/docs/latest/api/python/getting_started/.

Discussion Points:

How does Spark’s in-memory processing improve performance?
Why is PySpark suitable for Python developers?
What are the benefits of distributed computing for big data?
How can PySpark simplify ETL workflows?
What challenges might arise when scaling to larger datasets?

Week 2: Setting Up PySpark Environment

Introduction: Setting up a robust PySpark environment is essential for developing and running Spark applications. This week focuses on installing PySpark, configuring dependencies (e.g., Java, Python), and setting up tools like Jupyter Notebook for interactive development. The emphasis is on practical steps to create a local PySpark environment and verify its functionality with a simple data processing task.

Learning Objectives: By the end of this week, you will be able to:

Install and configure Java, Python, and PySpark dependencies.
Set up a local PySpark environment on your machine.
Use Jupyter Notebook for interactive PySpark development.
Verify the PySpark setup with a basic data processing task.
Troubleshoot common setup issues (e.g., Java version, path errors).

Scope: This week covers the setup of a PySpark environment, building on Week 1’s introduction to PySpark and Apache Spark. It prepares for Week 3’s exploration of RDD fundamentals by ensuring a functional development environment. The focus is on practical configuration and validation.

Background Information: A PySpark environment requires specific components:

Dependencies: Java: Spark requires Java 8, 11, or 17 (Java 8 recommended for compatibility). Python: Python 3.8+ for PySpark compatibility. PySpark: Installed via pip, includes Spark libraries. Optional Tools: Jupyter Notebook for interactive coding, findspark for easier PySpark initialization.
Environment Setup: Install dependencies and set environment variables (e.g., JAVA_HOME, SPARK_HOME). Use a local Spark cluster (standalone mode) for development. Configure Jupyter to run PySpark kernels.
Key Tools: pip: Installs PySpark and dependencies (pip install pyspark). Jupyter Notebook: Provides an interactive interface for coding and visualizing results. findspark: Simplifies PySpark imports by setting paths (pip install findspark).
Applications: Run PySpark scripts locally for development and testing. Prototype data processing pipelines before scaling to clusters.
Challenges: Java version mismatches (e.g., Java 21 causing errors). Incorrect environment variables (e.g., JAVA_HOME not set). PySpark installation issues due to network or dependency conflicts.

Hands-On Example:

Scenario: You’re a data analyst setting up a local PySpark environment to analyze a sales dataset. You’ll install dependencies, configure the environment, set up Jupyter Notebook, and run a PySpark script to verify the setup by counting records and filtering data.

Prerequisites:

Operating system: Windows, macOS, or Linux.
Internet connection for downloading dependencies.
Sample dataset (sales_data.csv) from Week 1:sale_id,product,price,region 1,Laptop,2000,North 2,Mouse,600,North 3,Keyboard,750,South 4,Tablet,1200,West

Step-by-Step Instructions:

Install Dependencies:

Java: Download Java 8 (JDK) from https://adoptium.net/temurin/releases/?version=8 or Oracle. Install, note installation path (e.g., /usr/lib/jvm/java-8-openjdk on Linux, C:\Program Files\Java\jdk1.8.0_391 on Windows). Verify: Open terminal, run java -version, expect output like:openjdk version "1.8.0_422"

Python: Download Python 3.8+ from https://www.python.org/downloads/. Install, ensure pip is included. Verify: Run python --version, expect Python 3.8.x or higher.

PySpark: In terminal, install PySpark:pip install pyspark==3.5.3 Verify: Run pip show pyspark, expect version 3.5.3.

Jupyter Notebook: Install: pip install jupyter. Verify: Run jupyter notebook, ensure browser opens.

findspark: Install: pip install findspark. Verify: Run python -c "import findspark", expect no errors.

Configure Environment Variables:

JAVA_HOME: Set JAVA_HOME to Java 8 path: Linux/macOS: Add to ~/.bashrc or ~/.zshrc:export JAVA_HOME=/usr/lib/jvm/java-8-openjdk export PATH=$JAVA_HOME/bin:$PATH Source file: source ~/.bashrc. Windows: Set via System Properties > Environment Variables: New variable: JAVA_HOME, value: C:\Program Files\Java\jdk1.8.0_391. Add %JAVA_HOME%\bin to PATH. Verify: Run echo $JAVA_HOME (Linux/macOS) or echo %JAVA_HOME% (Windows).

SPARK_HOME (optional, for advanced setups): PySpark’s pip installation includes Spark, so SPARK_HOME is not mandatory. If manually installing Spark, download from https://spark.apache.org/downloads.html, set SPARK_HOME.

Set Up Working Directory:

Create a directory (e.g., pyspark_course). Save sales_data.csv in the directory. Navigate to directory: cd pyspark_course.

Configure Jupyter for PySpark:

Create a Jupyter Notebook configuration: Run jupyter notebook --generate-config. Edit ~/.jupyter/jupyter_notebook_config.py, add:c.NotebookApp.notebook_dir = '/path/to/pyspark_course'

Launch Jupyter: jupyter notebook. Create a new notebook named “Week2_PySpark_Setup”.

Write and Run a PySpark Script:

In the notebook, add the following code to verify the PySpark setup:import findspark findspark.init() from pyspark.sql import SparkSession # Initialize SparkSession spark = SparkSession.builder \ .appName("SetupVerification") \ .config("spark.driver.memory", "2g") \ .getOrCreate() # Read CSV file df = spark.read.csv("sales_data.csv", header=True, inferSchema=True) # Show data df.show() # Count records record_count = df.count() print(f"Total records: {record_count}") # Filter South region south_sales = df.filter(df.region == "South") south_sales.show() south_count = south_sales.count() print(f"South region records: {south_count}") # Stop SparkSession spark.stop()

Explanation:

findspark.init(): Ensures PySpark is found in the Python path.

SparkSession: Configures a local Spark app with 2GB driver memory.

read.csv: Loads CSV with headers and schema inference.

filter: Selects rows where region is “South”.

show() and count(): Display and count data, demonstrating distributed processing.

Execute and Validate:

Check Spark UI: Run spark.sparkContext.uiWebUrl, open URL (e.g., http://localhost:4040). Verify jobs for show, count, and filter.

Validate: DataFrame matches CSV. Total records: 4. South records: 1 (Keyboard, $750).

If issues occur: Java Error: Ensure JAVA_HOME points to Java 8, not higher versions. File Not Found: Use absolute path for sales_data.csv. PySpark Import Error: Reinstall PySpark (pip install --force-reinstall pyspark). Memory Error: Reduce spark.driver.memory to 1g.

Clean Up: Stop SparkSession (included in code). Save and close Jupyter Notebook. Optionally, save notebook as Week2_PySpark_Setup.ipynb.

Output: A configured local PySpark environment with Java, Python, and PySpark. A working Jupyter Notebook running a PySpark script. Validated setup: Reads CSV, counts 4 records, filters 1 South record. Familiarity with troubleshooting setup issues.

PySpark Tips:

Use Java 8 to avoid compatibility issues.
Set JAVA_HOME correctly to prevent “Java not found” errors.
Run Jupyter in the project directory to simplify file access.
Use findspark for seamless PySpark imports.
Check PySpark docs: https://spark.apache.org/docs/latest/api/python/

Interpretation: This hands-on example demonstrates how to set up a local PySpark environment, install dependencies, and verify functionality with a simple data processing task. By troubleshooting common issues, you gain confidence in your setup, preparing for Week 3’s RDD fundamentals.

Supplemental Information: PySpark Installation: https://spark.apache.org/docs/latest/api/python/getting_started/install.html. Java Setup: https://adoptium.net/installation/. Jupyter with Spark: https://spark.apache.org/docs/latest/api/python/user_guide/python_packaging.html.

Discussion Points:

Why is Java 8 preferred for PySpark?
How does Jupyter Notebook enhance PySpark development?
What are the benefits of a local Spark environment?
How can environment variables impact PySpark?
What setup challenges might arise in a team setting?

Week 3: RDD Fundamentals & Transformations

Introduction: Resilient Distributed Datasets (RDDs) are the foundational data structure in Apache Spark, enabling distributed data processing with fault tolerance. This week focuses on understanding RDDs, creating them, and applying transformations to manipulate data. The emphasis is on practical usage through PySpark to process a sales dataset, demonstrating key RDD operations.

Learning Objectives: By the end of this week, you will be able to:

Explain the concept of RDDs and their role in Spark.
Create RDDs from various data sources (e.g., collections, files).
Apply common RDD transformations (e.g., map, filter, flatMap).
Understand lazy evaluation in Spark transformations.
Validate RDD operations and troubleshoot errors.

Scope: This week covers RDD fundamentals and transformations, building on Week 1’s introduction to PySpark and Week 2’s environment setup. It prepares for Week 4’s exploration of DataFrames and Spark SQL by establishing core distributed processing concepts. The focus is on hands-on RDD manipulation.

Background Information: RDDs are central to Spark’s distributed computing model:

RDD Overview: Resilient Distributed Dataset: A fault-tolerant collection of elements partitioned across a cluster. Key properties: Distributed: Data is split across nodes for parallel processing. Resilient: Lineage tracking allows recomputation of lost partitions. Immutable: Transformations create new RDDs, preserving the original. Stored in memory for fast access, with disk spillover if needed.
Creating RDDs: From collections: spark.sparkContext.parallelize([list]). From files: spark.sparkContext.textFile("file_path"). From other RDDs via transformations.
Transformations: Lazy operations that define a new RDD without immediate execution. Common transformations: map(func): Applies a function to each element. filter(func): Selects elements based on a condition. flatMap(func): Maps elements to a sequence and flattens results. distinct(): Removes duplicates. Lazy evaluation: Transformations are computed only when an action (e.g., count, collect) is called.
Actions: Trigger computation and return results (e.g., count(), collect(), take(n)).
Applications: Process large datasets (e.g., sales logs) by transforming and filtering. Prepare data for analysis or machine learning.
Challenges: Understanding lazy evaluation and lineage. Debugging transformation errors (e.g., incorrect function logic). Performance tuning for large RDDs.

Hands-On Example:

Scenario: You’re a data analyst using PySpark to analyze a sales dataset stored as a text file. You’ll create an RDD, apply transformations to clean and process the data, and use actions to retrieve results, demonstrating RDD fundamentals.

Prerequisites:

PySpark environment set up (from Week 2: Java 8, Python 3.8+, PySpark 3.5.3, Jupyter Notebook).
Sample dataset (sales_data.txt):1,Laptop,2000,North 2,Mouse,600,North 3,Keyboard,750,South 4,Tablet,1200,West 5,Laptop,2100,North

Save sales_data.txt in your working directory (e.g., pyspark_course).

Step-by-Step Instructions:

Set Up the Environment:

Navigate to your working directory: cd pyspark_course.

Launch Jupyter Notebook: jupyter notebook.

Create a new notebook named “Week3_RDD_Transformations”.

Verify PySpark installation:import pyspark print(pyspark.__version__) # Should print 3.5.3

Initialize Spark and Create an RDD:

Add the following code to initialize Spark and load the text file into an RDD:import findspark findspark.init() from pyspark.sql import SparkSession # Initialize SparkSession spark = SparkSession.builder \ .appName("RDDTransformations") \ .config("spark.driver.memory", "2g") \ .getOrCreate() # Access SparkContext sc = spark.sparkContext # Create RDD from text file rdd = sc.textFile("sales_data.txt") # Display first few lines print(rdd.take(3))

Run the cell, verify output:['1,Laptop,2000,North', '2,Mouse,600,North', '3,Keyboard,750,South']

Explanation:

SparkSession: Provides access to SparkContext via spark.sparkContext.

textFile: Creates an RDD where each line is a string.

take(3): Action to retrieve the first 3 lines.

Apply RDD Transformations:

Add code to parse the data, filter, and transform the RDD:# Parse lines into (sale_id, product, price, region) parsed_rdd = rdd.map(lambda line: line.split(",")) # Filter for North region north_rdd = parsed_rdd.filter(lambda x: x[3] == "North") # Map to (product, price) pairs product_price_rdd = north_rdd.map(lambda x: (x[1], int(x[2]))) # Get distinct products in North distinct_products = product_price_rdd.map(lambda x: x[0]).distinct() # Use flatMap to extract all regions (demonstrating flattening) all_regions = rdd.flatMap(lambda line: [line.split(",")[3]]) # Collect results print("Parsed RDD (first 2):", parsed_rdd.take(2)) print("North sales:", north_rdd.collect()) print("Product-price pairs (North):", product_price_rdd.collect()) print("Distinct North products:", distinct_products.collect()) print("All regions:", all_regions.collect())

Run, verify output:Parsed RDD (first 2): [['1', 'Laptop', '2000', 'North'], ['2', 'Mouse', '600', 'North']] North sales: [['1', 'Laptop', '2000', 'North'], ['2', 'Mouse', '600', 'North'], ['5', 'Laptop', '2100', 'North']] Product-price pairs (North): [('Laptop', 2000), ('Mouse', 600), ('Laptop', 2100)] Distinct North products: ['Laptop', 'Mouse'] All regions: ['North', 'North', 'South', 'West', 'North']

Explanation:

map: Splits each line into a list (e.g., ["1", "Laptop", "2000", "North"]).

filter: Selects rows where region is “North”.

map (again): Creates (product, price) tuples, converting price to integer.

distinct: Gets unique products in North.

flatMap: Extracts regions, flattening into a single list.

Transformations are lazy; collect and take trigger execution.

Perform an Action to Compute Results:

Add code to count total sales and sum North region prices:# Count total records total_count = parsed_rdd.count() print(f"Total sales records: {total_count}") # Sum prices for North region north_total = product_price_rdd.map(lambda x: x[1]).reduce(lambda x, y: x + y) print(f"Total North sales: ${north_total}")

Run, verify:Total sales records: 5 Total North sales: $4700

Explanation:

count: Action to count all records.

reduce: Action to sum prices (2000 + 600 + 2100 = 4700).

reduce runs on the cluster, aggregating values across partitions.

Validate and Troubleshoot:

Validation: Confirm parsed_rdd has 5 records, matching sales_data.txt. Verify North sales: 3 records (Laptop, Mouse, Laptop), total $4,700. Check distinct products: Laptop, Mouse (Laptop appears twice). Ensure regions include North (3), South (1), West (1).

Spark UI: Run spark.sparkContext.uiWebUrl, open URL (e.g., http://localhost:4040). Verify jobs for take, collect, count, reduce.

Troubleshooting: File Not Found: Ensure sales_data.txt is in the directory (use absolute path if needed). Index Error: Check map and filter logic (e.g., x[3] assumes 4 columns). Type Error: Ensure price conversion (int(x[2])) handles valid integers. No Output: Confirm actions (collect, count) are called to trigger transformations.

Clean Up:

Add to the notebook:# Stop SparkSession spark.stop()

Save notebook as “Week3_RDD_Transformations.ipynb”. Close Jupyter Notebook.

Output: A working PySpark script using RDDs to process a sales dataset. Proficiency in creating RDDs, applying transformations (map, filter, distinct, flatMap), and using actions (count, reduce). Validated results: 5 total records, 3 North sales ($4,700), 2 distinct North products. Understanding of lazy evaluation and Spark’s execution model.

PySpark Tips:

Use take(n) for quick debugging instead of collect() to avoid large data transfers.
Ensure transformation functions (e.g., lambda) handle edge cases (e.g., missing columns).
Check Spark UI to understand job execution and performance.
Chain transformations to minimize RDD creation.
Refer to RDD docs: https://spark.apache.org/docs/latest/rdd-programming-guide.html

Interpretation: This hands-on example demonstrates how to use PySpark RDDs to process a dataset, applying transformations to filter and transform data in a distributed manner. By validating results and exploring lazy evaluation, you build skills for Week 4’s DataFrames and Spark SQL.

Supplemental Information: RDD Programming Guide: https://spark.apache.org/docs/latest/rdd-programming-guide.html. PySpark RDD API: https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.html. Transformations vs. Actions: https://spark.apache.org/docs/latest/rdd-programming-guide.html#transformations.

Discussion Points:

How do RDDs enable fault-tolerant distributed processing?
Why is lazy evaluation beneficial for performance?
What are the trade-offs of using RDDs vs. DataFrames?
How can transformations optimize data pipelines?
What challenges arise when debugging RDD transformations?

Week 4: DataFrames & Spark SQL Basics

Introduction: PySpark DataFrames and Spark SQL provide a high-level, structured interface for distributed data processing, offering a more user-friendly alternative to RDDs. This week focuses on creating and manipulating DataFrames, performing operations like filtering and grouping, and using Spark SQL to query data. The emphasis is on practical usage through PySpark to analyze a sales dataset.

Learning Objectives: By the end of this week, you will be able to:

Understand the role of DataFrames and Spark SQL in PySpark.
Create DataFrames from various sources (e.g., CSV, RDDs).
Perform common DataFrame operations (e.g., filter, groupBy, join).
Write and execute Spark SQL queries.
Validate DataFrame and SQL results and troubleshoot errors.

Scope: This week covers DataFrames and Spark SQL basics, building on Week 1’s introduction, Week 2’s environment setup, and Week 3’s RDDs. It prepares for Week 5’s exploration of cluster architecture by introducing structured data processing. The focus is on hands-on DataFrame manipulation and SQL queries.

Background Information: DataFrames and Spark SQL simplify data processing in Spark:

DataFrames: Distributed collections of data organized into named columns, similar to tables in a relational database. Built on RDDs but provide a higher-level API with schema enforcement. Support operations like filtering, grouping, joining, and aggregations. Created from files (e.g., CSV, JSON), RDDs, or Spark SQL queries.
Spark SQL: Module for querying structured data using SQL syntax. Integrates with DataFrames: DataFrames can be registered as temporary views for SQL queries. Supports standard SQL operations (e.g., SELECT, WHERE, GROUP BY).
Key Operations: select(): Choose specific columns. filter(): Apply conditions to rows. groupBy(): Group data for aggregations (e.g., sum, count). join(): Combine DataFrames based on keys. createOrReplaceTempView(): Register DataFrame for SQL queries.
Applications: Analyze sales data by region or product. Build ETL pipelines with structured transformations. Query large datasets with familiar SQL syntax.
Challenges: Managing schema mismatches during DataFrame creation. Optimizing SQL queries for performance. Debugging errors in complex transformations.

Hands-On Example:

Scenario: You’re a data analyst using PySpark to analyze a sales dataset. You’ll create a DataFrame from a CSV file, perform transformations (filtering, grouping), and use Spark SQL to query the data, demonstrating structured processing.

Prerequisites:

PySpark environment set up (from Week 2: Java 8, Python 3.8+, PySpark 3.5.3, Jupyter Notebook).
Sample dataset (sales_data.csv):sale_id,product,price,region 1,Laptop,2000,North 2,Mouse,600,North 3,Keyboard,750,South 4,Tablet,1200,West 5,Laptop,2100,North

Save sales_data.csv in your working directory (e.g., pyspark_course).

Step-by-Step Instructions:

Set Up the Environment:

Navigate to your working directory: cd pyspark_course.

Launch Jupyter Notebook: jupyter notebook.

Create a new notebook named “Week4_DataFrames_SQL”.

Verify PySpark installation:import pyspark print(pyspark.__version__) # Should print 3.5.3

Create a DataFrame:

Add the following code to initialize Spark and load the CSV into a DataFrame:import findspark findspark.init() from pyspark.sql import SparkSession # Initialize SparkSession spark = SparkSession.builder \ .appName("DataFrameSQLBasics") \ .config("spark.driver.memory", "2g") \ .getOrCreate() # Read CSV into DataFrame df = spark.read.csv("sales_data.csv", header=True, inferSchema=True) # Display DataFrame df.show() df.printSchema()

Explanation:

read.csv: Loads CSV with headers and infers schema (e.g., price as integer).

show(): Displays the DataFrame.

printSchema(): Shows column names and data types.

Perform DataFrame Operations:

Add code to filter, select, and group data:from pyspark.sql.functions import col, sum, count # Select specific columns selected_df = df.select("product", "price", "region") selected_df.show() # Filter for price > 1000 high_price_df = df.filter(col("price") > 1000) high_price_df.show() # Group by region and calculate total sales and count region_summary = df.groupBy("region").agg( sum("price").alias("total_sales"), count("sale_id").alias("record_count") ) region_summary.show()

Run, verify output:+--------+-----+------+ | product|price|region| +--------+-----+------+ | Laptop| 2000| North| | Mouse| 600| North| |Keyboard| 750| South| | Tablet| 1200| West| | Laptop| 2100| North| +--------+-----+------+ +-------+-------+-----+------+ |sale_id|product|price|region| +-------+-------+-----+------+ | 1| Laptop| 2000| North| | 4| Tablet| 1200| West| | 5| Laptop| 2100| North| +-------+-------+-----+------+ +------+-----------+------------+ |region|total_sales|record_count| +------+-----------+------------+ | South| 750| 1| | West| 1200| 1| | North| 4700| 3| +------+-----------+------------+

Explanation:

select: Chooses specified columns.

filter: Selects rows where price > 1000 using col().

groupBy and agg: Groups by region, computes total sales and record count.

Use Spark SQL:

Add code to register the DataFrame as a temporary view and run SQL queries:# Register DataFrame as a temporary view df.createOrReplaceTempView("sales") # SQL query: Total sales by region region_sql = spark.sql(""" SELECT region, SUM(price) as total_sales, COUNT(sale_id) as record_count FROM sales GROUP BY region """) region_sql.show() # SQL query: Products with price > 1000 high_price_sql = spark.sql(""" SELECT product, price FROM sales WHERE price > 1000 ORDER BY price DESC """) high_price_sql.show()

Run, verify output:+------+-----------+------------+ |region|total_sales|record_count| +------+-----------+------------+ | South| 750| 1| | West| 1200| 1| | North| 4700| 3| +------+-----------+------------+ +-------+-----+ |product|price| +-------+-----+ | Laptop| 2100| | Laptop| 2000| | Tablet| 1200| +-------+-----+

Explanation:

createOrReplaceTempView: Registers DataFrame as “sales” for SQL queries.

spark.sql: Executes SQL queries, returning DataFrames.

Queries replicate DataFrame operations, showing SQL’s flexibility.

Validate and Troubleshoot:

Validation: Confirm DataFrame schema: sale_id (integer), product (string), price (integer), region (string). Verify high_price_df: 3 rows (Laptop, Tablet, Laptop). Check region_summary and region_sql: North ($4,700, 3), South ($750, 1), West ($1,200, 1). Ensure high_price_sql: Sorted by price (2100, 2000, 1200).

Spark UI: Run spark.sparkContext.uiWebUrl, open URL (e.g., http://localhost:4040). Verify jobs for show, filter, groupBy, and SQL queries.

Troubleshooting: Schema Error: Ensure inferSchema=True or define schema manually if CSV has inconsistencies. SQL Syntax Error: Check query syntax (e.g., correct column names). No Data: Verify sales_data.csv path and contents. Performance: Use df.cache() for repeated queries on small datasets (optional).

Clean Up:

Add to the notebook:# Stop SparkSession spark.stop()

Save notebook as “Week4_DataFrames_SQL.ipynb”. Close Jupyter Notebook.

Output: A working PySpark script using DataFrames and Spark SQL to process a sales dataset. Proficiency in creating DataFrames, performing operations (select, filter, groupBy), and running SQL queries. Validated results: 3 high-price products, region totals (North: $4,700, South: $750, West: $1,200). Understanding of structured data processing in PySpark.

PySpark Tips:

Use inferSchema=True for CSVs, but validate schema for production data.
Cache DataFrames (df.cache()) for repeated operations on small datasets.
Use col() in filters to avoid ambiguity with column names.
Write SQL queries for readability when sharing with SQL-familiar teams.
Refer to DataFrame docs: https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/dataframe.html

Interpretation: This hands-on example demonstrates how to use PySpark DataFrames and Spark SQL to process a sales dataset, offering a structured and SQL-like approach to distributed data analysis. By validating results, you build skills for Week 5’s cluster architecture exploration.

Supplemental Information: DataFrame API: https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/dataframe.html. Spark SQL Guide: https://spark.apache.org/docs/latest/sql-programming-guide.html. DataFrame Operations: https://spark.apache.org/docs/latest/api/python/user_guide/sql/index.html.

Discussion Points:

How do DataFrames simplify data processing compared to RDDs?
Why is Spark SQL valuable for analysts familiar with SQL?
What are the benefits of schema enforcement in DataFrames?
How can DataFrame operations optimize ETL pipelines?
What challenges arise when scaling DataFrame operations?

Week 5: PySpark Cluster Architecture

Introduction: Understanding PySpark’s cluster architecture is crucial for leveraging Spark’s distributed computing capabilities. This week focuses on the components of a Spark cluster (driver, executors, cluster manager), how they interact, and how to configure a local cluster for development. The emphasis is on practical usage, demonstrating how to run a PySpark job on a local cluster to process a sales dataset.

Learning Objectives: By the end of this week, you will be able to:

Explain Spark’s cluster architecture and its components.
Understand the roles of driver, executors, and cluster manager.
Configure a local PySpark cluster for development.
Run a PySpark job on a local cluster and monitor execution.
Troubleshoot cluster-related issues (e.g., resource allocation).

Scope: This week covers PySpark cluster architecture, building on Week 1’s introduction, Week 2’s environment setup, Week 3’s RDDs, and Week 4’s DataFrames and Spark SQL. It prepares for Week 6’s data ingestion by ensuring familiarity with distributed execution. The focus is on hands-on cluster configuration and job execution.

Background Information: Spark’s cluster architecture enables scalable data processing:

Components: Driver: Runs the main application, manages tasks, and coordinates with the cluster manager. Maintains the SparkContext/SparkSession. Executors: Worker nodes that execute tasks in parallel, store data partitions, and perform computations. Cluster Manager: Allocates resources (e.g., CPU, memory) across applications. Types include: Standalone: Spark’s built-in manager (used in this week). YARN: Hadoop’s resource manager. Mesos: General-purpose cluster manager. Kubernetes: Container orchestration platform.
Execution Flow: Driver submits a job, dividing it into tasks. Cluster manager assigns tasks to executors. Executors process data partitions and return results to the driver.
Local Mode: Simulates a cluster on a single machine (driver and executors run locally). Configured via master("local[*]") (uses all available cores).
Applications: Run distributed jobs for large-scale data processing. Test cluster behavior locally before deploying to production.
Challenges: Resource allocation (e.g., memory, cores) impacting performance. Debugging executor failures or driver bottlenecks. Configuring cluster settings for optimal execution.

Hands-On Example:

Scenario: You’re a data analyst configuring a local PySpark cluster to analyze a sales dataset. You’ll set up a SparkSession with a local cluster, process the data using DataFrame operations, and monitor the job via the Spark UI, demonstrating distributed execution.

Prerequisites:

PySpark environment set up (from Week 2: Java 8, Python 3.8+, PySpark 3.5.3, Jupyter Notebook).
Sample dataset (sales_data.csv):sale_id,product,price,region 1,Laptop,2000,North 2,Mouse,600,North 3,Keyboard,750,South 4,Tablet,1200,West 5,Laptop,2100,North

Save sales_data.csv in your working directory (e.g., pyspark_course).

Step-by-Step Instructions:

Set Up the Environment:

Navigate to your working directory: cd pyspark_course.

Launch Jupyter Notebook: jupyter notebook.

Create a new notebook named “Week5_Cluster_Architecture”.

Verify PySpark installation:import pyspark print(pyspark.__version__) # Should print 3.5.3

Configure a Local Cluster:

Add the following code to initialize a SparkSession with a local cluster:import findspark findspark.init() from pyspark.sql import SparkSession # Initialize SparkSession with local cluster spark = SparkSession.builder \ .appName("LocalClusterDemo") \ .master("local[*]") \ .config("spark.driver.memory", "2g") \ .config("spark.executor.memory", "1g") \ .config("spark.executor.cores", "1") \ .getOrCreate() # Display Spark configuration print("Spark Master:", spark.sparkContext.master) print("Application Name:", spark.sparkContext.appName)

Run the cell, verify output:Spark Master: local[*] Application Name: LocalClusterDemo

Explanation:

master("local[*]"): Runs Spark in local mode, using all available CPU cores.

spark.driver.memory: Allocates 2GB for the driver.

spark.executor.memory: Allocates 1GB per executor.

spark.executor.cores: Assigns 1 core per executor.

Load and Process Data:

Add code to load the CSV into a DataFrame and perform operations:from pyspark.sql.functions import col, sum, count # Read CSV into DataFrame df = spark.read.csv("sales_data.csv", header=True, inferSchema=True) # Show data df.show() # Cache DataFrame for repeated operations df.cache() # Group by region and calculate total sales and count region_summary = df.groupBy("region").agg( sum("price").alias("total_sales"), count("sale_id").alias("record_count") ) region_summary.show() # Filter for high-price sales (> $1000) high_price_df = df.filter(col("price") > 1000) high_price_df.show()

Run, verify output:+-------+--------+-----+------+ |sale_id|product |price|region| +-------+--------+-----+------+ | 1| Laptop| 2000| North| | 2| Mouse| 600| North| | 3|Keyboard| 750| South| | 4| Tablet| 1200| West| | 5| Laptop| 2100| North| +-------+--------+-----+------+ +------+-----------+------------+ |region|total_sales|record_count| +------+-----------+------------+ | South| 750| 1| | West| 1200| 1| | North| 4700| 3| +------+-----------+------------+ +-------+-------+-----+------+ |sale_id|product|price|region| +-------+-------+-----+------+ | 1| Laptop| 2000| North| | 4| Tablet| 1200| West| | 5| Laptop| 2100| North| +-------+-------+-----+------+

Explanation:

read.csv: Loads data with schema inference.

cache(): Stores DataFrame in memory for faster access in subsequent operations.

groupBy and agg: Computes total sales and counts per region.

filter: Selects high-price sales.

Monitor Execution via Spark UI:

Add code to access the Spark UI:print("Spark UI:", spark.sparkContext.uiWebUrl)

Run, note URL (e.g., http://localhost:4040).

Open URL in a browser, verify: Jobs: Check jobs for show, groupBy, and filter. Stages: Confirm tasks executed across local executors. Executors: Verify driver and executor details (e.g., 1g memory, 1 core).

Explanation:

Spark UI shows job execution, task distribution, and resource usage.

Local mode simulates multiple executors, showing parallel task execution.

Validate and Troubleshoot:

Validation: Confirm DataFrame: 5 rows, schema correct (e.g., price as integer). Verify region_summary: North ($4,700, 3), South ($750, 1), West ($1,200, 1). Check high_price_df: 3 rows (Laptop, Tablet, Laptop). Ensure Spark UI shows completed jobs without errors.

Troubleshooting: Memory Error: Reduce spark.driver.memory or spark.executor.memory if system resources are low. File Not Found: Use absolute path for sales_data.csv. UI Not Accessible: Ensure notebook is running and port 4040 is free. Slow Execution: Check if cache() is effective; small datasets may not show significant gains.

Clean Up:

Add to the notebook:# Unpersist cached DataFrame df.unpersist() # Stop SparkSession spark.stop()

Save notebook as “Week5_Cluster_Architecture.ipynb”. Close Jupyter Notebook.

Output: A configured local PySpark cluster running a DataFrame job. Proficiency in setting up a SparkSession with cluster settings and monitoring via Spark UI. Validated results: 5 records, region totals (North: $4,700, South: $750, West: $1,200), 3 high-price sales. Understanding of Spark’s driver-executor architecture.

PySpark Tips:

Use local[*] for development to maximize local resources.
Monitor Spark UI to understand task distribution and bottlenecks.
Cache DataFrames for repeated operations, but unpersist to free memory.
Adjust memory settings based on your machine’s capacity.
Refer to cluster docs: https://spark.apache.org/docs/latest/cluster-overview.html

Interpretation: This hands-on example demonstrates how to configure and use a local PySpark cluster to process a sales dataset, providing insight into Spark’s distributed architecture. By monitoring execution, you build skills for Week 6’s data ingestion and storage formats.

Supplemental Information: Cluster Overview: https://spark.apache.org/docs/latest/cluster-overview.html. Spark Configuration: https://spark.apache.org/docs/latest/configuration.html. Spark UI Guide: https://spark.apache.org/docs/latest/web-ui.html.

Discussion Points:

How does the driver coordinate with executors in a Spark job?
Why is a local cluster useful for development?
What are the trade-offs of different cluster managers?
How can resource allocation impact job performance?
What challenges arise in scaling from local to production clusters?

Week 6: Data Ingestion & Storage Formats

Introduction: Data ingestion and storage formats are critical for building scalable data pipelines in PySpark. This week focuses on ingesting data from various sources (e.g., CSV, JSON, Parquet) and working with different storage formats to optimize performance and compatibility. The emphasis is on practical usage, demonstrating how to read and write data in multiple formats using PySpark.

Learning Objectives: By the end of this week, you will be able to:

Ingest data from common sources (CSV, JSON, Parquet) into PySpark DataFrames.
Understand the characteristics of different storage formats (e.g., CSV, JSON, Parquet, Avro).
Write DataFrames to various storage formats.
Optimize data ingestion and storage for performance.
Validate ingested data and troubleshoot format-related issues.

Scope: This week covers data ingestion and storage formats, building on Week 1’s introduction, Week 2’s environment setup, Week 3’s RDDs, Week 4’s DataFrames and Spark SQL, and Week 5’s cluster architecture. It prepares for Week 7’s data manipulation by ensuring proficiency in handling diverse data sources and formats.

Background Information: PySpark supports flexible data ingestion and storage:

Data Ingestion: Sources: Files (CSV, JSON, Parquet, Avro, ORC), databases (via JDBC), streaming data. Methods: spark.read API (e.g., read.csv, read.json, read.parquet). Options: Schema inference, custom schemas, handling headers, and delimiters.
Storage Formats: CSV: Text-based, human-readable, supports headers, but lacks schema enforcement and compression. JSON: Semi-structured, supports nested data, but verbose and slower to process. Parquet: Columnar, optimized for big data, supports compression, schema evolution, and predicate pushdown. Avro: Row-based, schema-driven, supports evolution, suitable for streaming. ORC: Columnar, optimized for Hive, supports compression and indexing.
Key Considerations: Performance: Parquet and ORC are faster for large datasets due to columnar storage. Compatibility: CSV and JSON are widely supported but less efficient. Schema: Parquet and Avro enforce schemas, reducing errors.
Applications: Ingest sales data from CSV for initial analysis. Store processed data in Parquet for efficient querying. Read JSON logs for real-time analytics.
Challenges: Handling malformed data (e.g., missing fields, incorrect formats). Managing large files with limited memory. Ensuring compatibility across formats and tools.

Hands-On Example:

Scenario: You’re a data analyst tasked with ingesting sales data from CSV and JSON files into PySpark, processing it, and saving the results in Parquet and CSV formats. You’ll validate the data and optimize ingestion for performance.

Prerequisites:

PySpark environment set up (from Week 2: Java 8, Python 3.8+, PySpark 3.5.3, Jupyter Notebook).
Sample datasets: sales_data.csv:sale_id,product,price,region 1,Laptop,2000,North 2,Mouse,600,North 3,Keyboard,750,South 4,Tablet,1200,West 5,Laptop,2100,North new_sales.json:[ {"sale_id": 6, "product": "Monitor", "price": 1000, "region": "South"}, {"sale_id": 7, "product": "Headphones", "price": 300, "region": "West"} ]

Save both files in your working directory (e.g., pyspark_course).

Step-by-Step Instructions:

Set Up the Environment:

Navigate to your working directory: cd pyspark_course.

Launch Jupyter Notebook: jupyter notebook.

Create a new notebook named “Week6_Data_Ingestion”.

Verify PySpark installation:import pyspark print(pyspark.__version__) # Should print 3.5.3

Ingest Data from CSV and JSON:

Add the following code to initialize Spark and load data:import findspark findspark.init() from pyspark.sql import SparkSession # Initialize SparkSession spark = SparkSession.builder \ .appName("DataIngestion") \ .master("local[*]") \ .config("spark.driver.memory", "2g") \ .config("spark.executor.memory", "1g") \ .getOrCreate() # Read CSV csv_df = spark.read.csv("sales_data.csv", header=True, inferSchema=True) csv_df.show() csv_df.printSchema() # Read JSON json_df = spark.read.json("new_sales.json") json_df.show() json_df.printSchema()

Explanation:

read.csv: Loads CSV with headers and infers schema.

read.json: Loads JSON, automatically handling nested structures.

printSchema: Confirms data types (note JSON infers long for integers).

Combine and Process Data:

Add code to union the DataFrames and perform a simple aggregation:from pyspark.sql.functions import col, sum, count # Union CSV and JSON DataFrames combined_df = csv_df.union(json_df) combined_df.show() # Cache for performance combined_df.cache() # Group by region region_summary = combined_df.groupBy("region").agg( sum("price").alias("total_sales"), count("sale_id").alias("record_count") ) region_summary.show()

Run, verify output:+-------+----------+-----+------+ |sale_id| product |price|region| +-------+----------+-----+------+ | 1| Laptop| 2000| North| | 2| Mouse| 600| North| | 3| Keyboard| 750| South| | 4| Tablet| 1200| West| | 5| Laptop| 2100| North| | 6| Monitor| 1000| South| | 7|Headphones| 300| West| +-------+----------+-----+------+ +------+-----------+------------+ |region|total_sales|record_count| +------+-----------+------------+ | South| 1750| 2| | West| 1500| 2| | North| 4700| 3| +------+-----------+------------+

Explanation:

union: Combines DataFrames (schemas must match).

cache: Stores DataFrame in memory for faster access.

groupBy and agg: Computes total sales and counts per region.

Write Data to Parquet and CSV:

Add code to save the combined DataFrame in Parquet and CSV formats:# Write to Parquet combined_df.write.parquet("output/sales_parquet", mode="overwrite") print("Saved to Parquet") # Write to CSV combined_df.write.csv("output/sales_csv", header=True, mode="overwrite") print("Saved to CSV") # Read back Parquet to verify parquet_df = spark.read.parquet("output/sales_parquet") parquet_df.show()

Run, verify output:Saved to Parquet Saved to CSV +-------+----------+-----+------+ |sale_id| product |price|region| +-------+----------+-----+------+ | 1| Laptop| 2000| North| | 2| Mouse| 600| North| | 3| Keyboard| 750| South| | 4| Tablet| 1200| West| | 5| Laptop| 2100| North| | 6| Monitor| 1000| South| | 7|Headphones| 300| West| +-------+----------+-----+------+

Explanation:

write.parquet: Saves in columnar Parquet format, optimized for big data.

write.csv: Saves in text-based CSV format, widely compatible.

mode="overwrite": Overwrites existing output directories.

read.parquet: Verifies Parquet data integrity.

Validate and Troubleshoot:

Validation: Confirm combined_df: 7 rows, correct schema (despite JSON’s long types, union succeeds). Verify region_summary: North ($4,700, 3), South ($1,750, 2), West ($1,500, 2). Check output directories: output/sales_parquet: Contains part files (e.g., part-00000-*.parquet). output/sales_csv: Contains CSV part files with headers. Confirm parquet_df matches combined_df.

Spark UI: Run spark.sparkContext.uiWebUrl, open URL (e.g., http://localhost:4040). Verify jobs for show, union, groupBy, write, and read.

Troubleshooting: Schema Mismatch: Ensure CSV and JSON schemas align (e.g., same column names). File Not Found: Verify file paths (use absolute paths if needed). Write Error: Check write permissions in output directory; delete old output if needed. Slow Read/Write: Use Parquet for large datasets; CSV is slower.

Clean Up:

Add to the notebook:# Unpersist cached DataFrame combined_df.unpersist() # Stop SparkSession spark.stop()

Save notebook as “Week6_Data_Ingestion.ipynb”. Close Jupyter Notebook. Optionally, delete output directory: rm -r output.

Output: A working PySpark script ingesting CSV and JSON data, processing it, and saving in Parquet and CSV. Proficiency in reading/writing multiple formats and optimizing ingestion. Validated results: 7 records, region totals (North: $4,700, South: $1,750, West: $1,500). Understanding of storage format trade-offs.

PySpark Tips:

Use Parquet for large-scale data due to compression and speed.
Validate schemas before union to avoid errors.
Cache DataFrames for repeated operations, but unpersist to free memory.
Use inferSchema=True for small datasets, but define schemas for production.
Refer to I/O docs: https://spark.apache.org/docs/latest/sql-data-sources.html

Interpretation: This hands-on example demonstrates how to ingest data from CSV and JSON, process it, and save in Parquet and CSV formats using PySpark. By validating and optimizing data handling, you build skills for Week 7’s data manipulation.

Supplemental Information: Data Sources: https://spark.apache.org/docs/latest/sql-data-sources.html. Parquet Format: https://spark.apache.org/docs/latest/sql-data-sources-parquet.html. JSON Format: https://spark.apache.org/docs/latest/sql-data-sources-json.html.

Discussion Points:

Why is Parquet preferred for big data processing?
How do storage formats impact pipeline performance?
What are the challenges of ingesting semi-structured data like JSON?
How can schema inference improve or complicate ingestion?
What role does data ingestion play in ETL pipelines?

Week 7: Basic Data Manipulation in PySpark

Introduction: Data manipulation is a core component of data processing in PySpark, enabling transformations, cleaning, and enrichment of datasets. This week focuses on using PySpark DataFrame operations to perform common data manipulation tasks such as filtering, joining, aggregating, and handling missing data. The emphasis is on practical usage, demonstrating how to manipulate a sales dataset to derive insights.

Learning Objectives: By the end of this week, you will be able to:

Perform filtering and sorting operations on DataFrames.
Join multiple DataFrames to combine data.
Aggregate data using group-by operations.
Handle missing or null values in datasets.
Validate manipulated data and troubleshoot common issues.

Scope: This week covers basic data manipulation in PySpark, building on Week 1’s introduction, Week 2’s environment setup, Week 3’s RDDs, Week 4’s DataFrames and Spark SQL, Week 5’s cluster architecture, and Week 6’s data ingestion. It prepares for Week 8’s exploration of SparkContext and SparkSession by ensuring proficiency in DataFrame manipulation.

Background Information: PySpark DataFrames provide a powerful API for data manipulation:

Key Operations: Filtering: Select rows based on conditions (filter, where). Sorting: Order data by columns (orderBy). Joining: Combine DataFrames based on keys (join). Aggregation: Summarize data using group-by and functions like sum, count, avg. Handling Nulls: Identify, fill, or drop missing values (isNull, fillna, dropna).
Functions: Use pyspark.sql.functions for operations like col, when, lit to manipulate columns.
Performance: Lazy evaluation optimizes execution plans. Caching improves performance for repeated operations.
Applications: Clean and transform sales data for reporting. Join datasets (e.g., sales and customer data) for analysis. Aggregate data to compute metrics like total sales by region.
Challenges: Managing skewed joins or large datasets. Handling inconsistent or missing data. Debugging complex transformation pipelines.

Hands-On Example:

Scenario: You’re a data analyst tasked with manipulating a sales dataset and a customer dataset in PySpark. You’ll filter, join, aggregate, and handle missing data to generate a report on sales performance by region and customer segment.

Prerequisites:

PySpark environment set up (from Week 2: Java 8, Python 3.8+, PySpark 3.5.3, Jupyter Notebook).
Sample datasets: sales_data.csv:sale_id,product,price,region,customer_id 1,Laptop,2000,North,101 2,Mouse,600,North,102 3,Keyboard,750,South,103 4,Tablet,1200,West,104 5,Laptop,2100,North,101 6,Monitor,1000,South, customer_data.csv:customer_id,segment 101,Enterprise 102,Consumer 103,SMB 104,Enterprise 105,Consumer

Save both files in your working directory (e.g., pyspark_course).

Step-by-Step Instructions:

Set Up the Environment:

Navigate to your working directory: cd pyspark_course.

Launch Jupyter Notebook: jupyter notebook.

Create a new notebook named “Week7_Data_Manipulation”.

Verify PySpark installation:import pyspark print(pyspark.__version__) # Should print 3.5.3

Load and Inspect Data:

Add the following code to initialize Spark and load the datasets:import findspark findspark.init() from pyspark.sql import SparkSession # Initialize SparkSession spark = SparkSession.builder \ .appName("DataManipulation") \ .master("local[*]") \ .config("spark.driver.memory", "2g") \ .config("spark.executor.memory", "1g") \ .getOrCreate() # Read sales and customer data sales_df = spark.read.csv("sales_data.csv", header=True, inferSchema=True) customer_df = spark.read.csv("customer_data.csv", header=True, inferSchema=True) # Show data sales_df.show() customer_df.show() # Print schemas sales_df.printSchema() customer_df.printSchema()

Run, verify output:+-------+--------+-----+------+-----------+ |sale_id| product|price|region|customer_id| +-------+--------+-----+------+-----------+ | 1| Laptop| 2000| North| 101| | 2| Mouse| 600| North| 102| | 3|Keyboard| 750| South| 103| | 4| Tablet| 1200| West| 104| | 5| Laptop| 2100| North| 101| | 6| Monitor| 1000| South| null| +-------+--------+-----+------+-----------+ root |-- sale_id: integer (nullable = true) |-- product: string (nullable = true) |-- price: integer (nullable = true) |-- region: string (nullable = true) |-- customer_id: integer (nullable = true) +-----------+---------+ |customer_id| segment| +-----------+---------+ | 101|Enterprise| | 102| Consumer| | 103| SMB| | 104|Enterprise| | 105| Consumer| +-----------+---------+ root |-- customer_id: integer (nullable = true) |-- segment: string (nullable = true)

Explanation:

read.csv: Loads CSVs with schema inference.

show: Displays data; note null customer_id in sales (sale_id 6).

printSchema: Confirms data types.

Filter and Sort Data:

Add code to filter high-value sales and sort by price:from pyspark.sql.functions import col # Filter sales with price > 1000 high_value_sales = sales_df.filter(col("price") > 1000) high_value_sales.show() # Sort by price descending sorted_sales = high_value_sales.orderBy(col("price").desc()) sorted_sales.show()

Run, verify output:+-------+-------+-----+------+-----------+ |sale_id|product|price|region|customer_id| +-------+-------+-----+------+-----------+ | 1| Laptop| 2000| North| 101| | 4| Tablet| 1200| West| 104| | 5| Laptop| 2100| North| 101| +-------+-------+-----+------+-----------+ +-------+-------+-----+------+-----------+ |sale_id|product|price|region|customer_id| +-------+-------+-----+------+-----------+ | 5| Laptop| 2100| North| 101| | 1| Laptop| 2000| North| 101| | 4| Tablet| 1200| West| 104| +-------+-------+-----+------+-----------+

Explanation:

filter: Selects rows where price > 1000.

orderBy: Sorts by price in descending order.

Join DataFrames:

Add code to join sales and customer data:# Left join sales with customer data joined_df = sales_df.join(customer_df, "customer_id", "left") joined_df.show()

Run, verify output:+-----------+-------+--------+-----+------+---------+ |customer_id|sale_id| product|price|region| segment| +-----------+-------+--------+-----+------+---------+ | 101| 1| Laptop| 2000| North|Enterprise| | 101| 5| Laptop| 2100| North|Enterprise| | 102| 2| Mouse| 600| North| Consumer| | 103| 3|Keyboard| 750| South| SMB| | 104| 4| Tablet| 1200| West|Enterprise| | null| 6| Monitor| 1000| South| null| +-----------+-------+--------+-----+------+---------+

Explanation:

join: Performs a left join on customer_id, retaining all sales rows.

Sale_id 6 has null customer_id and segment.

Handle Missing Data:

Add code to handle null values:# Fill null segment with 'Unknown' filled_df = joined_df.fillna({"segment": "Unknown"}) filled_df.show() # Drop rows with null customer_id cleaned_df = filled_df.dropna(subset=["customer_id"]) cleaned_df.show()

Run, verify output:+-----------+-------+--------+-----+------+---------+ |customer_id|sale_id| product|price|region| segment| +-----------+-------+--------+-----+------+---------+ | 101| 1| Laptop| 2000| North|Enterprise| | 101| 5| Laptop| 2100| North|Enterprise| | 102| 2| Mouse| 600| North| Consumer| | 103| 3|Keyboard| 750| South| SMB| | 104| 4| Tablet| 1200| West|Enterprise| | null| 6| Monitor| 1000| South| Unknown| +-----------+-------+--------+-----+------+---------+ +-----------+-------+--------+-----+------+---------+ |customer_id|sale_id| product|price|region| segment| +-----------+-------+--------+-----+------+---------+ | 101| 1| Laptop| 2000| North|Enterprise| | 101| 5| Laptop| 2100| North|Enterprise| | 102| 2| Mouse| 600| North| Consumer| | 103| 3|Keyboard| 750| South| SMB| | 104| 4| Tablet| 1200| West|Enterprise| +-----------+-------+--------+-----+------+---------+

Explanation:

fillna: Replaces null segment with “Unknown”.

dropna: Removes rows with null customer_id (sale_id 6).

Aggregate Data:

Add code to aggregate sales by region and segment:from pyspark.sql.functions import sum, count # Cache cleaned DataFrame cleaned_df.cache() # Group by region and segment summary_df = cleaned_df.groupBy("region", "segment").agg( sum("price").alias("total_sales"), count("sale_id").alias("record_count") ) summary_df.show()

Run, verify output:+------+---------+-----------+------------+ |region| segment|total_sales|record_count| +------+---------+-----------+------------+ | North|Enterprise| 4100| 2| | North| Consumer| 600| 1| | South| SMB| 750| 1| | West|Enterprise| 1200| 1| +------+---------+-----------+------------+

Explanation:

cache: Stores DataFrame in memory for faster aggregation.

groupBy and agg: Computes total sales and counts by region and segment.

Validate and Troubleshoot:

Validation: Confirm joined_df: 6 rows, nulls in sale_id 6. Verify cleaned_df: 5 rows, no null customer_id. Check summary_df: Matches expected totals (e.g., North-Enterprise: $4,100 from two Laptops). Ensure schema consistency across operations.

Spark UI: Run spark.sparkContext.uiWebUrl, open URL (e.g., http://localhost:4040). Verify jobs for show, filter, join, fillna, dropna, and groupBy.

Troubleshooting: Join Errors: Ensure customer_id exists in both DataFrames; check join type. Null Issues: Verify fillna and dropna target correct columns. Performance: Use cache for repeated operations; unpersist when done. Data Loss: Check if dropna removes too many rows.

Clean Up:

Add to the notebook:# Unpersist cached DataFrame cleaned_df.unpersist() # Stop SparkSession spark.stop()

Save notebook as “Week7_Data_Manipulation.ipynb”. Close Jupyter Notebook.

Output: A working PySpark script manipulating sales and customer data. Proficiency in filtering, joining, handling nulls, and aggregating DataFrames. Validated results: 5 cleaned records, aggregated totals by region and segment. Understanding of data manipulation workflows.

PySpark Tips:

Use col() for explicit column references in filters and joins.
Cache DataFrames for complex pipelines, but unpersist to free memory.
Validate join results to avoid data duplication or loss.
Use where as an alias for filter for SQL-like syntax.
Refer to DataFrame docs: https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/dataframe.html

Interpretation: This hands-on example demonstrates how to manipulate data in PySpark using filtering, joining, null handling, and aggregation. By validating results, you build skills for Week 8’s exploration of SparkContext and SparkSession.

Supplemental Information: DataFrame Operations: https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/dataframe.html. Handling Nulls: https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.fillna.html. Joins: https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.join.html.

Discussion Points:

How do DataFrame operations streamline data preparation?
Why is handling nulls important in data pipelines?
What are the trade-offs of different join types?
How can caching improve manipulation performance?
What challenges arise in manipulating large datasets?

Week 8: Working with SparkContext & SparkSession

Introduction: SparkContext and SparkSession are the entry points to PySpark, enabling interaction with Spark’s distributed computing engine. This week focuses on understanding their roles, configuring them, and using them to perform RDD and DataFrame operations. The emphasis is on practical usage, demonstrating how to initialize and manage these components to process a sales dataset.

Learning Objectives: By the end of this week, you will be able to:

Explain the roles of SparkContext and SparkSession in PySpark.
Initialize and configure SparkContext and SparkSession for different tasks.
Use SparkContext for RDD operations and SparkSession for DataFrame/SQL operations.
Manage Spark configurations to optimize performance.
Validate operations and troubleshoot context-related issues.

Scope: This week covers SparkContext and SparkSession, building on Week 1’s introduction, Week 2’s environment setup, Week 3’s RDDs, Week 4’s DataFrames and Spark SQL, Week 5’s cluster architecture, Week 6’s data ingestion, and Week 7’s data manipulation. It prepares for Week 9’s version control by ensuring proficiency in Spark’s core APIs.

Background Information: SparkContext and SparkSession are foundational to PySpark applications:

SparkContext: The original entry point for Spark (pre-Spark 2.0). Manages low-level operations, such as creating RDDs and setting configurations. Accessed via spark.sparkContext in modern PySpark. Used for RDD-based tasks and cluster resource management.
SparkSession: Introduced in Spark 2.0 as a unified entry point. Combines SparkContext, SQLContext, and HiveContext functionalities. Supports DataFrames, Spark SQL, and dataset APIs. Preferred for most modern PySpark applications.
Key Configurations: appName: Sets the application name (e.g., .appName("MyApp")). master: Specifies the cluster manager (e.g., local[*], yarn). spark.driver.memory: Allocates driver memory. spark.executor.memory: Allocates executor memory. spark.sql.shuffle.partitions: Sets partitions for SQL/DataFrame operations.
Applications: Initialize Spark for distributed data processing. Switch between RDD and DataFrame APIs based on task requirements. Optimize resource usage for performance.
Challenges: Managing multiple contexts in a single application. Avoiding resource conflicts (e.g., memory overruns). Debugging configuration errors or context initialization failures.

Hands-On Example:

Scenario: You’re a data analyst using PySpark to analyze a sales dataset. You’ll initialize SparkSession and SparkContext, perform RDD operations to count unique products, and use DataFrame operations to summarize sales by region, demonstrating the interplay between these components.

Prerequisites:

PySpark environment set up (from Week 2: Java 8, Python 3.8+, PySpark 3.5.3, Jupyter Notebook).
Sample dataset (sales_data.csv):sale_id,product,price,region 1,Laptop,2000,North 2,Mouse,600,North 3,Keyboard,750,South 4,Tablet,1200,West 5,Laptop,2100,North

Save sales_data.csv in your working directory (e.g., pyspark_course).

Step-by-Step Instructions:

Set Up the Environment:

Navigate to your working directory: cd pyspark_course.

Launch Jupyter Notebook: jupyter notebook.

Create a new notebook named “Week8_SparkContext_Session”.

Verify PySpark installation:import pyspark print(pyspark.__version__) # Should print 3.5.3

Initialize SparkSession and SparkContext:

Add the following code to initialize SparkSession and access SparkContext:import findspark findspark.init() from pyspark.sql import SparkSession # Initialize SparkSession with custom configurations spark = SparkSession.builder \ .appName("ContextSessionDemo") \ .master("local[*]") \ .config("spark.driver.memory", "2g") \ .config("spark.executor.memory", "1g") \ .config("spark.sql.shuffle.partitions", "4") \ .getOrCreate() # Access SparkContext sc = spark.sparkContext # Display configurations print("Application Name:", spark.sparkContext.appName) print("Master:", spark.sparkContext.master) print("Spark Version:", spark.version) print("Shuffle Partitions:", spark.conf.get("spark.sql.shuffle.partitions"))

Run, verify output:Application Name: ContextSessionDemo Master: local[*] Spark Version: 3.5.3 Shuffle Partitions: 4

Explanation:

SparkSession.builder: Configures the session with app name, local cluster, and resource settings.

config: Sets shuffle partitions to 4 for DataFrame operations.

spark.sparkContext: Accesses the underlying SparkContext.

Output confirms the configuration.

Use SparkContext for RDD Operations:

Add code to create an RDD and count unique products:# Load CSV as RDD rdd = sc.textFile("sales_data.csv") # Skip header and parse product column header = rdd.first() product_rdd = rdd.filter(lambda line: line != header) \ .map(lambda line: line.split(",")[1]) # Count unique products unique_products = product_rdd.distinct().collect() print("Unique Products:", unique_products) print("Number of Unique Products:", len(unique_products))

Run, verify output:Unique Products: ['Laptop', 'Mouse', 'Keyboard', 'Tablet'] Number of Unique Products: 4

Explanation:

textFile: Loads CSV as an RDD of lines.

filter: Skips the header row.

map: Extracts the product column (index 1).

distinct: Gets unique products.

collect: Returns results to the driver.

Use SparkSession for DataFrame Operations:

Add code to load the CSV as a DataFrame and summarize sales by region:from pyspark.sql.functions import col, sum, count # Load CSV as DataFrame df = spark.read.csv("sales_data.csv", header=True, inferSchema=True) df.show() # Cache DataFrame df.cache() # Summarize sales by region region_summary = df.groupBy("region").agg( sum("price").alias("total_sales"), count("sale_id").alias("record_count") ) region_summary.show()

Run, verify output:+-------+--------+-----+------+ |sale_id| product|price|region| +-------+--------+-----+------+ | 1| Laptop| 2000| North| | 2| Mouse| 600| North| | 3|Keyboard| 750| South| | 4| Tablet| 1200| West| | 5| Laptop| 2100| North| +-------+--------+-----+------+ +------+-----------+------------+ |region|total_sales|record_count| +------+-----------+------------+ | South| 750| 1| | West| 1200| 1| | North| 4700| 3| +------+-----------+------------+

Explanation:

read.csv: Loads CSV into a DataFrame with schema inference.

cache: Stores DataFrame in memory for faster access.

groupBy and agg: Computes total sales and counts per region.

Combine RDD and DataFrame Operations:

Add code to convert the RDD to a DataFrame and perform a SQL query:# Convert product RDD to DataFrame product_df = product_rdd.map(lambda x: (x,)).toDF(["product"]) product_df.show() # Register DataFrame as a temporary view region_summary.createOrReplaceTempView("region_sales") # Run SQL query high_sales_regions = spark.sql(""" SELECT region, total_sales FROM region_sales WHERE total_sales > 1000 ORDER BY total_sales DESC """) high_sales_regions.show()

Explanation:

toDF: Converts RDD to DataFrame with a single column.

createOrReplaceTempView: Registers DataFrame for SQL queries.

spark.sql: Filters regions with total sales > $1,000.

Validate and Troubleshoot:

Validation: Confirm product_rdd: 4 unique products (Laptop, Mouse, Keyboard, Tablet). Verify region_summary: North ($4,700, 3), South ($750, 1), West ($1,200, 1). Check high_sales_regions: North ($4,700), West ($1,200). Ensure product_df includes all products, including duplicates.

Spark UI: Run spark.sparkContext.uiWebUrl, open URL (e.g., http://localhost:4040). Verify jobs for collect, show, groupBy, and sql.

Troubleshooting: RDD Errors: Ensure map and split handle CSV correctly (e.g., check delimiters). DataFrame Errors: Verify schema with printSchema if groupBy fails. Context Issues: Ensure only one SparkSession is active; stop previous sessions. Memory Issues: Reduce spark.driver.memory if out-of-memory errors occur.

Clean Up:

Add to the notebook:# Unpersist cached DataFrame df.unpersist() # Stop SparkSession spark.stop()

Save notebook as “Week8_SparkContext_Session.ipynb”. Close Jupyter Notebook.

Output: A working PySpark script using SparkContext for RDD operations and SparkSession for DataFrame/SQL tasks. Proficiency in initializing and configuring Spark’s entry points. Validated results: 4 unique products, region totals (North: $4,700, South: $750, West: $1,200), high-sales regions. Understanding of SparkContext and SparkSession interplay.

PySpark Tips:

Use SparkSession for most tasks; access SparkContext only for RDDs.
Set configurations at initialization to avoid runtime errors.
Monitor Spark UI to verify job execution and resource usage.
Stop SparkSession to free resources after use.
Refer to SparkSession docs: https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/spark_session.html

Interpretation: This hands-on example demonstrates how to use SparkContext and SparkSession to perform mixed RDD and DataFrame operations on a sales dataset. By validating results, you build skills for Week 9’s version control in PySpark projects.

Supplemental Information: SparkSession API: https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/spark_session.html. SparkContext API: https://spark.apache.org/docs/latest/api/python/reference/pyspark/context.html. Configuration Guide: https://spark.apache.org/docs/latest/configuration.html.

Discussion Points:

How does SparkSession simplify PySpark development compared to SparkContext?
Why configure resources at the start of a Spark application?
What are the benefits of combining RDD and DataFrame operations?
How can misconfigured contexts impact performance?
What challenges arise when managing multiple Spark applications?

Week 9: Version Control for PySpark Projects

Introduction: Version control is essential for managing PySpark projects, enabling collaboration, tracking changes, and ensuring reproducibility. This week focuses on using Git, a popular version control system, to manage PySpark code, configurations, and data pipelines. The emphasis is on practical usage, demonstrating how to set up a Git repository, commit changes, and collaborate on a PySpark project.

Learning Objectives: By the end of this week, you will be able to:

Understand the role of version control in PySpark projects.
Set up a Git repository for a PySpark project.
Commit, branch, and merge changes using Git.
Collaborate on PySpark code using remote repositories (e.g., GitHub).
Validate version control workflows and troubleshoot Git issues.

Scope: This week covers version control with Git, building on Week 1’s introduction, Week 2’s environment setup, Week 3’s RDDs, Week 4’s DataFrames and Spark SQL, Week 5’s cluster architecture, Week 6’s data ingestion, Week 7’s data manipulation, and Week 8’s SparkContext and SparkSession. It prepares for Week 10’s Spark ecosystem by ensuring proficiency in managing project code.

Background Information: Git is widely used for version control in data engineering projects:

Git Basics: Repository: A directory containing project files and their version history. Commit: A snapshot of changes to files. Branch: A parallel version of the repository for isolated development. Merge: Combines changes from multiple branches. Remote Repository: A hosted repository (e.g., GitHub, GitLab) for collaboration.
PySpark Project Structure: Scripts: Python files (.py) or Jupyter notebooks (.ipynb) containing PySpark code. Data: Small sample datasets (e.g., CSV) or references to external data. Configurations: Environment settings or Spark configurations. .gitignore: Specifies files to exclude (e.g., large data files, Spark checkpoints).
Key Commands: git init: Initializes a repository. git add: Stages changes for commit. git commit: Saves staged changes. git branch: Manages branches. git merge: Merges branches. git push/pull: Syncs with remote repositories.
Applications: Track changes to PySpark scripts for data pipelines. Collaborate with team members on shared codebases. Revert to previous versions if errors occur.
Challenges: Managing large data files (avoided by using external storage). Resolving merge conflicts in collaborative projects. Ensuring consistent environments across collaborators.

Hands-On Example:

Scenario: You’re a data analyst developing a PySpark script to analyze a sales dataset. You’ll create a Git repository, commit changes to the script, create a branch for a new feature, merge it, and push to a remote repository (e.g., GitHub) to simulate collaboration.

Prerequisites:

PySpark environment set up (from Week 2: Java 8, Python 3.8+, PySpark 3.5.3, Jupyter Notebook).
Git installed (download from https://git-scm.com/).
GitHub account and a new repository created (e.g., pyspark-project).
Sample dataset (sales_data.csv):sale_id,product,price,region 1,Laptop,2000,North 2,Mouse,600,North 3,Keyboard,750,South 4,Tablet,1200,West 5,Laptop,2100,North

Save sales_data.csv in your working directory (e.g., pyspark_course).

Step-by-Step Instructions:

Set Up the Environment:

Navigate to your working directory: cd pyspark_course.

Verify Git installation:git --version

Expect output like git version 2.x.x.

Configure Git (if not already set):git config --global user.name "Your Name" git config --global user.email "your.email@example.com"

Initialize a Git Repository:

Create a project directory and initialize Git:mkdir pyspark_project cd pyspark_project git init

Create a .gitignore file to exclude unnecessary files:echo -e "# PySpark\n*.parquet\n*.csv\n__pycache__/\n*.pyc\n# Jupyter\n*.ipynb_checkpoints/\n# Spark\nspark-warehouse/\nmetastore_db/\n" > .gitignore

Copy sales_data.csv to the directory:cp ../sales_data.csv .

Create and Commit Initial PySpark Script:

Create a Python script (sales_analysis.py) with basic PySpark code:from pyspark.sql import SparkSession from pyspark.sql.functions import col, sum, count # Initialize SparkSession spark = SparkSession.builder \ .appName("SalesAnalysis") \ .master("local[*]") \ .config("spark.driver.memory", "2g") \ .getOrCreate() # Load data df = spark.read.csv("sales_data.csv", header=True, inferSchema=True) df.show() # Summarize by region region_summary = df.groupBy("region").agg( sum("price").alias("total_sales"), count("sale_id").alias("record_count") ) region_summary.show() # Stop SparkSession spark.stop()

Stage and commit the files:git add .gitignore sales_analysis.py git commit -m "Initial commit: Add sales analysis script and .gitignore"

Set Up Remote Repository:

Link to the GitHub repository:git remote add origin https://github.com/your-username/pyspark-project.git

Push the initial commit:git push -u origin main

Verify the repository on GitHub: Check that .gitignore and sales_analysis.py are uploaded.

Create a Feature Branch:

Create a branch for a new feature (e.g., adding product count):git checkout -b add-product-count

Modify sales_analysis.py to include product count:from pyspark.sql import SparkSession from pyspark.sql.functions import col, sum, count # Initialize SparkSession spark = SparkSession.builder \ .appName("SalesAnalysis") \ .master("local[*]") \ .config("spark.driver.memory", "2g") \ .getOrCreate() # Load data df = spark.read.csv("sales_data.csv", header=True, inferSchema=True) df.show() # Summarize by region region_summary = df.groupBy("region").agg( sum("price").alias("total_sales"), count("sale_id").alias("record_count") ) region_summary.show() # Count unique products unique_products = df.select("product").distinct().count() print(f"Number of unique products: {unique_products}") # Stop SparkSession spark.stop()

Test the script:python sales_analysis.py

Verify output:+-------+--------+-----+------+ |sale_id| product|price|region| +-------+--------+-----+------+ | 1| Laptop| 2000| North| | 2| Mouse| 600| North| | 3|Keyboard| 750| South| | 4| Tablet| 1200| West| | 5| Laptop| 2100| North| +-------+--------+-----+------+ +------+-----------+------------+ |region|total_sales|record_count| +------+-----------+------------+ | South| 750| 1| | West| 1200| 1| | North| 4700| 3| +------+-----------+------------+ Number of unique products: 4

Commit the changes:git add sales_analysis.py git commit -m "Add unique product count feature"

Merge the Feature Branch:

Switch to the main branch and merge:git checkout main git merge add-product-count

Push the updated main branch:git push origin main

Verify on GitHub: Ensure sales_analysis.py includes the product count feature.

Simulate Collaboration:

Create a new branch for a collaborator’s change (e.g., filtering high-value sales):git checkout -b high-value-filter

Update sales_analysis.py to add a filter:from pyspark.sql import SparkSession from pyspark.sql.functions import col, sum, count # Initialize SparkSession spark = SparkSession.builder \ .appName("SalesAnalysis") \ .master("local[*]") \ .config("spark.driver.memory", "2g") \ .getOrCreate() # Load data df = spark.read.csv("sales_data.csv", header=True, inferSchema=True) df.show() # Summarize by region region_summary = df.groupBy("region").agg( sum("price").alias("total_sales"), count("sale_id").alias("record_count") ) region_summary.show() # Count unique products unique_products = df.select("product").distinct().count() print(f"Number of unique products: {unique_products}") # Filter high-value sales (> $1000) high_value_df = df.filter(col("price") > 1000) high_value_df.show() # Stop SparkSession spark.stop()

Test the script:python sales_analysis.py

Verify output includes:+-------+-------+-----+------+ |sale_id|product|price|region| +-------+-------+-----+------+ | 1| Laptop| 2000| North| | 4| Tablet| 1200| West| | 5| Laptop| 2100| North| +-------+-------+-----+------+

Commit and push the branch:git add sales_analysis.py git commit -m "Add high-value sales filter" git push origin high-value-filter

On GitHub, create a pull request for high-value-filter to main, review, and merge it.

Pull the updated main branch locally:git checkout main git pull origin main

Validate and Troubleshoot:

Validation: Confirm repository contains .gitignore, sales_analysis.py, and sales_data.csv. Verify commits: Run git log --oneline to see commit history. Check final sales_analysis.py: Includes region summary, unique product count, and high-value filter. Test the script to ensure all features work (region totals, 4 unique products, 3 high-value sales).

Troubleshooting: Merge Conflicts: Resolve by editing conflicting lines and committing. Push Errors: Ensure correct remote URL (git remote -v) and authentication. File Exclusions: Verify .gitignore excludes large or temporary files. Script Errors: Check PySpark code for syntax or path issues.

Clean Up:

Delete local feature branches:git branch -d add-product-count high-value-filter

Ensure repository is clean: git status.

Keep the project directory for future weeks.

Output: A Git repository managing a PySpark project with a functional script. Proficiency in Git commands (init, add, commit, branch, merge, push). Validated workflow: Script analyzes sales data with region summaries, unique product count (4), and high-value sales filter (3 rows). Experience with collaborative version control via GitHub.

PySpark and Git Tips:

Use .gitignore to exclude large data files and Spark artifacts.
Commit frequently with descriptive messages to track changes.
Use branches for features to avoid disrupting the main branch.
Test scripts before committing to ensure functionality.
Refer to Git docs: https://git-scm.com/doc

Interpretation: This hands-on example demonstrates how to use Git for version control in a PySpark project, from initializing a repository to collaborating via GitHub. By validating the workflow, you build skills for Week 10’s exploration of the Spark ecosystem.

Supplemental Information: Git Documentation: https://git-scm.com/doc. GitHub Guides: https://docs.github.com/en/get-started. PySpark Project Structure: https://spark.apache.org/docs/latest/api/python/user_guide/python_packaging.html.

Discussion Points:

How does version control improve PySpark project management?
Why use branches for feature development?
What are the benefits of remote repositories like GitHub?
How can .gitignore optimize repository management?
What challenges arise in collaborative PySpark projects?

Week 10: Introduction to Spark Ecosystem

Introduction: The Spark ecosystem extends beyond core data processing, offering tools for SQL queries, machine learning, streaming, and graph processing. This week introduces key components of the Spark ecosystem—Spark SQL, MLlib, Spark Streaming, and GraphX—and demonstrates their practical application in PySpark. The emphasis is on practical usage, showing how to integrate these components to analyze a sales dataset and simulate real-time data processing.

Learning Objectives: By the end of this week, you will be able to:

Understand the components of the Spark ecosystem and their use cases.
Use Spark SQL for advanced querying.
Apply MLlib for basic machine learning tasks.
Implement Spark Streaming for real-time data processing.
Validate ecosystem integrations and troubleshoot issues.

Scope: This week introduces the Spark ecosystem, building on Week 1’s introduction, Week 2’s environment setup, Week 3’s RDDs, Week 4’s DataFrames and Spark SQL, Week 5’s cluster architecture, Week 6’s data ingestion, Week 7’s data manipulation, Week 8’s SparkContext and SparkSession, and Week 9’s version control. It concludes the course by showcasing Spark’s versatility.

Background Information: The Spark ecosystem provides a unified platform for diverse data processing tasks:

Spark SQL: Enables structured querying using SQL or DataFrame APIs. Supports complex queries, joins, and window functions. Integrates with external data sources (e.g., Hive, JDBC).
MLlib: Spark’s machine learning library for distributed ML. Offers algorithms for classification, regression, clustering, and more. Scales to large datasets with Spark’s distributed computing.
Spark Streaming: Processes real-time data streams (e.g., logs, sensor data). Uses micro-batches for near-real-time processing. Integrates with sources like Kafka, Flume, or file systems.
GraphX: Processes graph data (e.g., social networks) with distributed algorithms. Less common in PySpark due to limited Python support.
Applications: Run SQL queries for sales reporting. Build predictive models with MLlib (e.g., sales forecasting). Process real-time sales transactions with Streaming.
Challenges: Managing dependencies for MLlib or Streaming. Optimizing streaming pipelines for latency. Ensuring data consistency across components.

Hands-On Example:

Scenario: You’re a data analyst using PySpark to analyze a sales dataset. You’ll use Spark SQL to generate a sales report, MLlib to cluster products by price, and Spark Streaming to simulate real-time sales data processing, demonstrating the Spark ecosystem’s capabilities.

Prerequisites:

PySpark environment set up (from Week 2: Java 8, Python 3.8+, PySpark 3.5.3, Jupyter Notebook).
Sample datasets: sales_data.csv (batch data):sale_id,product,price,region 1,Laptop,2000,North 2,Mouse,600,North 3,Keyboard,750,South 4,Tablet,1200,West 5,Laptop,2100,North realtime_sales directory with CSV files for streaming (e.g., sale1.csv, sale2.csv): sale1.csv:sale_id,product,price,region 6,Monitor,1000,South sale2.csv:sale_id,product,price,region 7,Headphones,300,West

Save sales_data.csv and create a realtime_sales directory with the streaming CSVs in your working directory (e.g., pyspark_course).

Install additional dependencies:pip install numpy

Step-by-Step Instructions:

Set Up the Environment:

Navigate to your working directory: cd pyspark_course.

Launch Jupyter Notebook: jupyter notebook.

Create a new notebook named “Week10_Spark_Ecosystem”.

Verify PySpark installation:import pyspark print(pyspark.__version__) # Should print 3.5.3

Initialize SparkSession:

Add the following code to initialize SparkSession:import findspark findspark.init() from pyspark.sql import SparkSession # Initialize SparkSession spark = SparkSession.builder \ .appName("SparkEcosystemDemo") \ .master("local[*]") \ .config("spark.driver.memory", "2g") \ .config("spark.executor.memory", "1g") \ .getOrCreate() # Verify SparkSession print("Spark Version:", spark.version)

Run, verify output:Spark Version: 3.5.3

Use Spark SQL for Advanced Querying:

Add code to load batch data and run a complex SQL query:# Load batch data df = spark.read.csv("sales_data.csv", header=True, inferSchema=True) df.show() # Register as temporary view df.createOrReplaceTempView("sales") # SQL query: Rank products by price within each region ranked_sales = spark.sql(""" SELECT region, product, price, RANK() OVER (PARTITION BY region ORDER BY price DESC) as price_rank FROM sales ORDER BY region, price_rank """) ranked_sales.show()

Run, verify output:+-------+--------+-----+------+ |sale_id| product|price|region| +-------+--------+-----+------+ | 1| Laptop| 2000| North| | 2| Mouse| 600| North| | 3|Keyboard| 750| South| | 4| Tablet| 1200| West| | 5| Laptop| 2100| North| +-------+--------+-----+------+ +------+--------+-----+----------+ |region| product|price|price_rank| +------+--------+-----+----------+ | North| Laptop| 2100| 1| | North| Laptop| 2000| 2| | North| Mouse| 600| 3| | South|Keyboard| 750| 1| | West| Tablet| 1200| 1| +------+--------+-----+----------+

Explanation:

createOrReplaceTempView: Registers DataFrame for SQL queries.

RANK() OVER: Window function ranks products by price within each region.

Output shows top-priced products per region.

Use MLlib for Clustering:

Add code to cluster products by price using K-means:from pyspark.ml.feature import VectorAssembler from pyspark.ml.clustering import KMeans # Prepare data for clustering assembler = VectorAssembler(inputCols=["price"], outputCol="features") feature_df = assembler.transform(df) # Train K-means model kmeans = KMeans(k=2, seed=1) model = kmeans.fit(feature_df) # Predict clusters clustered_df = model.transform(feature_df) clustered_df.select("product", "price", "prediction").show()

Run, verify output (clusters may vary based on initialization):+--------+-----+----------+ | product|price|prediction| +--------+-----+----------+ | Laptop| 2000| 0| | Mouse| 600| 1| |Keyboard| 750| 1| | Tablet| 1200| 0| | Laptop| 2100| 0| +--------+-----+----------+

Explanation:

VectorAssembler: Converts price into a feature vector.

KMeans: Clusters products into 2 groups (e.g., high vs. low price).

prediction: Cluster ID (0 or 1); high-price products (Laptop, Tablet) likely in one cluster, low-price (Mouse, Keyboard) in another.

Use Spark Streaming for Real-Time Processing:

Add code to process streaming sales data:from pyspark.sql.types import StructType, StructField, StringType, IntegerType # Define schema for streaming data schema = StructType([ StructField("sale_id", IntegerType(), True), StructField("product", StringType(), True), StructField("price", IntegerType(), True), StructField("region", StringType(), True) ]) # Read streaming data streaming_df = spark.readStream \ .schema(schema) \ .csv("realtime_sales") # Aggregate total sales by region stream_summary = streaming_df.groupBy("region").agg( sum("price").alias("total_sales") ) # Write stream to console query = stream_summary.writeStream \ .outputMode("complete") \ .format("console") \ .start() # Wait for streaming to process (simulate by adding files to realtime_sales) query.awaitTermination(10) # Run for 10 seconds

Before running, ensure sale1.csv and sale2.csv are in realtime_sales.

Run the cell, verify console output (may vary based on timing):+------+-----------+ |region|total_sales| +------+-----------+ | South| 1000| | West| 300| +------+-----------+

Explanation:

readStream: Reads CSVs from realtime_sales as a stream.

schema: Ensures correct data types.

groupBy: Aggregates total sales by region.

writeStream: Outputs results to console in “complete” mode (full table).

Simulate streaming by adding files to realtime_sales during execution.

Validate and Troubleshoot:

Validation: Confirm ranked_sales: Correct ranking within regions (e.g., Laptop $2,100 ranks 1 in North). Verify clustered_df: Clusters group high-price (e.g., Laptop, Tablet) and low-price (e.g., Mouse, Keyboard) products. Check stream_summary: Reflects streaming data (South: $1,000, West: $300). Ensure schemas match across components.

Spark UI: Run spark.sparkContext.uiWebUrl, open URL (e.g., http://localhost:4040). Verify jobs for SQL, MLlib, and streaming; check streaming tab for micro-batch details.

Troubleshooting: SQL Errors: Verify column names and window function syntax. MLlib Errors: Ensure features column is numeric; check for nulls. Streaming Issues: Confirm realtime_sales path and schema; ensure files are added during execution. Memory Issues: Reduce spark.driver.memory if errors occur.

Clean Up:

Add to the notebook:# Stop streaming query query.stop() # Stop SparkSession spark.stop()

Save notebook as “Week10_Spark_Ecosystem.ipynb”. Close Jupyter Notebook. Optionally, delete realtime_sales directory: rm -r realtime_sales.

Output: A working PySpark script integrating Spark SQL, MLlib, and Spark Streaming. Proficiency in using Spark ecosystem components for querying, clustering, and streaming. Validated results: Ranked sales by region, 2 price clusters, streaming totals (South: $1,000, West: $300). Understanding of Spark’s unified ecosystem.

PySpark Tips:

Use explicit schemas for streaming to avoid inference errors.
Cache DataFrames for complex SQL or MLlib tasks, but unpersist afterward.
Monitor Spark UI’s streaming tab for real-time job details.
Start with small datasets for MLlib to tune models before scaling.
Refer to ecosystem docs: https://spark.apache.org/docs/latest/

Interpretation: This hands-on example demonstrates how to leverage the Spark ecosystem—Spark SQL for querying, MLlib for clustering, and Spark Streaming for real-time processing—in a PySpark project. By validating results, you complete the course with a comprehensive understanding of Spark’s capabilities.

Supplemental Information: Spark SQL Guide: https://spark.apache.org/docs/latest/sql-programming-guide.html. MLlib Guide: https://spark.apache.org/docs/latest/ml-guide.html. Spark Streaming Guide: https://spark.apache.org/docs/latest/streaming-programming-guide.html.

Discussion Points:

How does the Spark ecosystem unify diverse data tasks?
Why is Spark SQL suitable for analysts familiar with SQL?
What are the benefits of MLlib for large-scale machine learning?
How does Spark Streaming enable real-time analytics?
What challenges arise when integrating multiple Spark components?