Key Concepts of PySpark Core Programming

1. Resilient Distributed Dataset (RDD)

RDD is the fundamental data structure in PySpark. It is an immutable, distributed collection of objects that can be processed in parallel.

Features of RDD:

Immutable: Once created, cannot be changed.

Distributed: Stored across multiple nodes.

Lazy Evaluation: Operations are executed only when an action is triggered.

Fault Tolerance: Can recover lost partitions using lineage.

RDD Creation Methods:

from pyspark.sql import SparkSession 
spark = SparkSession.builder.appName("RDDExample").getOrCreate() 
sc = spark.sparkContext # SparkContext is needed for RDD operations 
# Creating RDD from a list rdd1 = sc.parallelize([1, 2, 3, 4, 5]) 
# Creating RDD from a file rdd2 = sc.textFile("sample.txt")

2. RDD Transformations

Transformations create a new RDD from an existing one. They are lazy, meaning they are executed only when an action is performed.

• Common Transformations:
  • map(): Applies a function to each element.
  • filter(): Filters elements based on a condition.
  • flatMap(): Similar to map() but flattens the results.
  • reduceByKey(): Groups data by key and applies an aggregation function.
• • distinct(): Removes duplicates.
  Example:
• rdd = sc.parallelize([1, 2, 3, 4, 5])
• # Applying map transformation squared_rdd = rdd.map(lambda x: x * x)
• print(squared_rdd.collect()) # Output: [1, 4, 9, 16, 25]

3. RDD Actions

Actions trigger the execution of transformations and return results.

Common Actions:

Example: numbers = sc.parallelize([1, 2, 3, 4, 5]) 
total = numbers.reduce(lambda a, b: a + b)
 print(total) # Output: 15

4. RDD Persistence (Caching and Checkpointing)

cache(): Stores RDD in memory to speed up computation.

persist(storageLevel): Stores RDD in memory or disk with a specific storage level.

Example: rdd = sc.parallelize(range(1, 10000)) rdd.cache() # Cache the RDD in memory

5. Pair RDDs (Key-Value RDDs)

Useful for working with key-value pairs.

Common Operations:

summed = pairs.reduceByKey(lambda a, b: a + b)

print(summed.collect()) # Output: [('a', 4), ('b', 2)]

PySpark Core vs. PySpark SQL

While PySpark Core uses RDDs, PySpark SQL works with DataFrames (more optimized and user-friendly). However, RDDs are still useful for low-level control over distributed processing.

When to Use PySpark Core (RDDs)?

• When working with low-level transformations and actions.
• When handling unstructured or semi-structured data.
• When requiring fine-grained control over execution.
• When implementing custom partitioning.

For most tasks, PySpark SQL (DataFrames) is preferred due to better performance and optimizations.

PySpark RDD Transformations & Complex Use Cases

RDD transformations are lazy operations that return a new RDD without modifying the original one. They are only executed when an action (e.g., collect(), count()) is called.

✅ 1. map(func)

Description:

Applies a function to each element and returns a new RDD.

Complex Use Case:

Extracting specific fields from a large dataset.

rdd = sc.parallelize(["1,John,30", "2,Alice,25", "3,Bob,40"])
rdd_transformed = rdd.map(lambda x: (x.split(",")[1], int(x.split(",")[2])))
print(rdd_transformed.collect())  
# Output: [('John', 30), ('Alice', 25), ('Bob', 40)]

✅ 2. flatMap(func)

Description:

Similar to map(), but it flattens the results.

Complex Use Case:

Splitting sentences into words (tokenization for NLP).

rdd = sc.parallelize(["Hello World", "Spark is fast"])
rdd_flat = rdd.flatMap(lambda x: x.split(" "))
print(rdd_flat.collect())  
# Output: ['Hello', 'World', 'Spark', 'is', 'fast']

✅ 3. filter(func)

Description:

Filters elements based on a function.

Complex Use Case:

Filtering large log files for error messages.

rdd = sc.parallelize(["INFO: System started", "ERROR: Disk failure", "INFO: User login"])
rdd_errors = rdd.filter(lambda x: "ERROR" in x)
print(rdd_errors.collect())  
# Output: ['ERROR: Disk failure']

✅ 4. distinct()

Description:

Removes duplicate elements.

Complex Use Case:

Removing duplicate user logins.

rdd = sc.parallelize(["user1", "user2", "user1", "user3"])
rdd_distinct = rdd.distinct()
print(rdd_distinct.collect())  
# Output: ['user1', 'user2', 'user3']

✅ 5. groupByKey() (Pair RDD)

Description:

Groups values by key.

Complex Use Case:

Grouping scores of students.

rdd = sc.parallelize([("Alice", 85), ("Bob", 90), ("Alice", 95)])
rdd_grouped = rdd.groupByKey().mapValues(list)
print(rdd_grouped.collect())  
# Output: [('Alice', [85, 95]), ('Bob', [90])]

🔴 Problem: Shuffles all data → Use reduceByKey() if aggregation is needed.

✅ 6. reduceByKey(func) (Pair RDD)

Description:

Aggregates values by key using a function.

Complex Use Case:

Finding total sales per product.

rdd = sc.parallelize([("apple", 5), ("banana", 10), ("apple", 7)])
rdd_reduced = rdd.reduceByKey(lambda x, y: x + y)
print(rdd_reduced.collect())  
# Output: [('apple', 12), ('banana', 10)]

✅ Efficient: Works at the map-side, reducing shuffle.

✅ 7. sortByKey(ascending=True/False)

Description:

Sorts key-value pairs by key.

Complex Use Case:

Sorting student scores in descending order.

rdd = sc.parallelize([(85, "Alice"), (90, "Bob"), (75, "Charlie")])
rdd_sorted = rdd.sortByKey(ascending=False)
print(rdd_sorted.collect())  
# Output: [(90, 'Bob'), (85, 'Alice'), (75, 'Charlie')]

✅ 8. mapValues(func) (Pair RDD)

Description:

Applies a function only on values.

Complex Use Case:

Applying a grade scale to student scores.

rdd = sc.parallelize([("Alice", 85), ("Bob", 90)])
rdd_scaled = rdd.mapValues(lambda x: "A" if x > 80 else "B")
print(rdd_scaled.collect())  
# Output: [('Alice', 'A'), ('Bob', 'A')]

✅ 9. join() (Pair RDD)

Description:

Performs an inner join between two RDDs.

Complex Use Case:

Joining student names with their scores.

rdd_names = sc.parallelize([("101", "Alice"), ("102", "Bob")])
rdd_scores = sc.parallelize([("101", 85), ("102", 90)])

rdd_joined = rdd_names.join(rdd_scores)
print(rdd_joined.collect())  
# Output: [('101', ('Alice', 85)), ('102', ('Bob', 90))]

🔴 Can cause data shuffling. Use broadcast variables if one RDD is small.

✅ 10. coalesce(numPartitions)

Description:

Reduces the number of partitions.

Complex Use Case:

Optimizing small datasets after filtering.

rdd = sc.parallelize(range(100), numSlices=10)
rdd_coalesced = rdd.coalesce(5)
print(rdd_coalesced.getNumPartitions())  
# Output: 5

✅ Efficient: Works without shuffle (unless shuffle=True).

✅ 11. repartition(numPartitions)

Description:

Increases or decreases the number of partitions with shuffling.

Complex Use Case:

Rebalancing partitions before expensive operations.

rdd = sc.parallelize(range(100), numSlices=5)
rdd_repartitioned = rdd.repartition(10)
print(rdd_repartitioned.getNumPartitions())  
# Output: 10

🔴 Expensive: Causes full data shuffle.

✅ 12. union(rdd)

Description:

Merges two RDDs without removing duplicates.

Complex Use Case:

Combining logs from different sources.

rdd1 = sc.parallelize(["log1", "log2"])
rdd2 = sc.parallelize(["log3", "log4"])

rdd_combined = rdd1.union(rdd2)
print(rdd_combined.collect())  
# Output: ['log1', 'log2', 'log3', 'log4']

✅ 13. intersection(rdd)

Description:

Finds common elements between two RDDs.

Complex Use Case:

Finding common customers in two datasets.

rdd1 = sc.parallelize(["Alice", "Bob", "Charlie"])
rdd2 = sc.parallelize(["Bob", "Charlie", "David"])

rdd_common = rdd1.intersection(rdd2)
print(rdd_common.collect())  
# Output: ['Charlie', 'Bob']

✅ 14. zip(rdd)

Description:

Merges two RDDs element-wise.

Complex Use Case:

Pairing product IDs with names.

rdd1 = sc.parallelize([1, 2, 3])
rdd2 = sc.parallelize(["Apple", "Banana", "Cherry"])

rdd_zipped = rdd1.zip(rdd2)
print(rdd_zipped.collect())  
# Output: [(1, 'Apple'), (2, 'Banana'), (3, 'Cherry')]

🔴 Requires same number of partitions and elements.

🚀 Summary

🔥 Use Cases for Transforming Spark DataFrames to RDDs and Performing Operations

In PySpark, we often transform DataFrames (DFs) to RDDs when we need fine-grained control over data, perform custom transformations that are difficult in DataFrames, or leverage RDD-specific operations. However, converting a DF to an RDD loses schema information, so it’s recommended only when necessary.

✅ 1. Complex Custom Transformations (Not Available in Spark SQL)

Use Case: Applying a custom transformation that requires stateful row-wise operations

If a transformation is too complex for DataFrame APIs (e.g., multi-row aggregation, state tracking), using RDDs can be more efficient.

🔹 Example: Convert a DataFrame of transactions into an RDD, and track cumulative revenue per user.

from pyspark.sql import SparkSession, Row

spark = SparkSession.builder.appName("DF_to_RDD").getOrCreate()

# Creating a sample DataFrame
data = [("Alice", 100), ("Bob", 200), ("Alice", 300), ("Bob", 100)]
df = spark.createDataFrame(data, ["user", "amount"])

# Convert to RDD
rdd = df.rdd.map(lambda row: (row.user, row.amount))

# Custom transformation: cumulative sum per user
from collections import defaultdict
user_revenue = defaultdict(int)

def cumulative_sum(record):
    user, amount = record
    user_revenue[user] += amount
    return (user, user_revenue[user])

result_rdd = rdd.map(cumulative_sum)
print(result_rdd.collect())

# Output (May vary due to parallel execution):
# [('Alice', 100), ('Bob', 200), ('Alice', 400), ('Bob', 300)]

✅ Why RDD? This approach allows tracking cumulative state across records, which is hard in Spark SQL.

✅ 2. Handling JSON/Complex Data Structures in a Row

Use Case: When a DataFrame has nested JSON structures that need row-wise parsing.

🔹 Example: Extract fields from a JSON column using RDDs.

from pyspark.sql import Row
import json

data = [Row(id=1, json_data='{"name": "Alice", "age": 25}'),
        Row(id=2, json_data='{"name": "Bob", "age": 30}')]

df = spark.createDataFrame(data)

# Convert to RDD
rdd = df.rdd.map(lambda row: (row.id, json.loads(row.json_data)["name"], json.loads(row.json_data)["age"]))

# Convert back to DataFrame with new schema
df_transformed = rdd.toDF(["id", "name", "age"])
df_transformed.show()

✅ Why RDD? It allows working with nested structures in a way that DataFrame SQL functions don’t easily support.

✅ 3. Custom Sorting and Partitioning for Optimization

Use Case: When we need fine-grained control over how data is partitioned or sorted.

🔹 Example: Sorting data by multiple custom keys before converting it back to a DataFrame.

data = [("Alice", 25, 50000), ("Bob", 30, 60000), ("Charlie", 22, 55000)]
df = spark.createDataFrame(data, ["name", "age", "salary"])

# Convert to RDD and sort by age first, then by salary
sorted_rdd = df.rdd.sortBy(lambda row: (row.age, row.salary), ascending=True)

# Convert back to DataFrame
sorted_df = sorted_rdd.toDF(["name", "age", "salary"])
sorted_df.show()

✅ Why RDD? DataFrame .orderBy() can be costly; RDD sorting allows finer control over key-based ordering.

✅ 4. Row-Level Data Anonymization & Masking

Use Case: Anonymizing sensitive data before processing it further.

🔹 Example: Mask email addresses before writing to a new table.

data = [("Alice", "alice@example.com"), ("Bob", "bob@example.com")]
df = spark.createDataFrame(data, ["name", "email"])

# Convert to RDD and anonymize email
rdd = df.rdd.map(lambda row: (row.name, row.email.split("@")[0] + "@****.com"))

# Convert back to DataFrame
masked_df = rdd.toDF(["name", "masked_email"])
masked_df.show()

✅ Why RDD? Some transformations, like string manipulation, are easier with Python’s built-in functions.

✅ 5. Handling Heterogeneous Data (Different Column Types per Row)

Use Case: When DataFrame rows have different schemas or formats dynamically.

🔹 Example: Processing a dataset with mixed types and dynamically inferring schema.

data = [(1, "Alice", 25), (2, "Bob", "unknown"), (3, "Charlie", 30)]
df = spark.createDataFrame(data, ["id", "name", "age"])

# Convert to RDD and fix mixed data types
rdd = df.rdd.map(lambda row: (row.id, row.name, int(row.age) if row.age.isdigit() else None))

# Convert back to DataFrame
fixed_df = rdd.toDF(["id", "name", "age"])
fixed_df.show()

✅ Why RDD? Unlike DataFrame operations, RDD allows handling heterogeneous row structures dynamically.

✅ 6. Custom Grouping & Aggregation (Beyond Built-in SQL Functions)

Use Case: Implementing custom aggregations that aren’t directly supported by Spark SQL.

🔹 Example: Computing a weighted average manually.

data = [("Alice", 80, 4), ("Alice", 90, 6), ("Bob", 70, 3), ("Bob", 60, 2)]
df = spark.createDataFrame(data, ["name", "score", "weight"])

# Convert to RDD
rdd = df.rdd.map(lambda row: (row.name, (row.score * row.weight, row.weight)))

# Compute weighted average
aggregated_rdd = rdd.reduceByKey(lambda a, b: (a[0] + b[0], a[1] + b[1]))
weighted_avg_rdd = aggregated_rdd.map(lambda x: (x[0], x[1][0] / x[1][1]))

# Convert back to DataFrame
result_df = weighted_avg_rdd.toDF(["name", "weighted_avg"])
result_df.show()

✅ Why RDD? Custom aggregations require more control than SQL provides.

🚀 When Should You Convert DataFrame to RDD?

🔥 Key Takeaways

✔ RDDs provide more flexibility but lose schema information.
✔ Convert to RDD only if necessary—DataFrame optimizations are usually better.
✔ Best use cases: Custom transformations, JSON handling, complex aggregations, stateful row operations.

Yes, there is a difference in how RDDs and DataFrames are created and executed in PySpark. Let’s break it down.

1️⃣ RDD Creation & Execution

• RDDs require SparkContext (sc) for creation.
• They use low-level transformations and actions.
• They are lazy, meaning they execute only when an action is called.

RDD Creation Examples

from pyspark.sql import SparkSession

# Create SparkSession
spark = SparkSession.builder.appName("RDDExample").getOrCreate()

# Get SparkContext from SparkSession
sc = spark.sparkContext  

# Create RDD from a list
rdd = sc.parallelize([1, 2, 3, 4, 5])

# Apply transformation (lazy, does not execute yet)
rdd_squared = rdd.map(lambda x: x * x)

# Action triggers execution
print(rdd_squared.collect())  # Output: [1, 4, 9, 16, 25]

RDD Execution Flow

1. RDD is created → Stored as a lineage graph (DAG, not executed yet).
2. Transformations are applied → Still lazy.
3. Action is triggered → Spark executes the DAG and computes results.

2️⃣ DataFrame Creation & Execution

• DataFrames are created using SparkSession (not SparkContext).
• They are optimized internally using the Catalyst Optimizer.
• They are lazily evaluated like RDDs, but optimizations (like predicate pushdown) make them faster.

DataFrame Creation Examples

# Create SparkSession (No need for SparkContext explicitly)
spark = SparkSession.builder.appName("DFExample").getOrCreate()

# Create DataFrame from a list
df = spark.createDataFrame([(1, "Alice"), (2, "Bob")], ["id", "name"])

# Apply transformation (lazy)
df_filtered = df.filter(df.id > 1)

# Action triggers execution
df_filtered.show()

DataFrame Execution Flow

1. DataFrame is created → Internally optimized into a logical plan.
2. Transformations are applied → Still lazy, but optimized by the Catalyst Optimizer.
3. Action is triggered (show(), collect(), etc.) → Spark executes the optimized DAG and computes results.

3️⃣ Key Differences in Execution

4️⃣ When is RDD Explicitly Needed?

• When dealing with low-level transformations.
• When performing custom partitioning or optimizations.
• When working with unstructured or complex data.

Otherwise, DataFrames are preferred because they are faster and optimized.

🔍 Serialization and Deserialization in RDDs vs. DataFrames

In Apache Spark, serialization and deserialization (SerDe) play a critical role in how data is processed across a cluster. The major performance difference between RDDs and DataFrames arises from how they handle serialization and deserialization (SerDe) when distributing data.

🔥 1. Why Does RDD Require Serialization/Deserialization?

• RDD (Resilient Distributed Dataset) is a low-level API where each record is a raw Python object (e.g., list, tuple, dictionary).
• Since Spark runs on a distributed cluster, data must be serialized (converted into a byte stream) before being sent to worker nodes, and then deserialized (converted back into an object) when received.
• Serialization is expensive in terms of CPU and memory usage.

Example of RDD Serialization Overhead

Let’s create an RDD and analyze serialization:

import pickle
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("RDD_Serialization").getOrCreate()

data = [("Alice", 25), ("Bob", 30), ("Charlie", 35)]
rdd = spark.sparkContext.parallelize(data)

# Simulating serialization manually
serialized_data = rdd.map(lambda x: pickle.dumps(x))  # Converting Python object to byte stream
deserialized_data = serialized_data.map(lambda x: pickle.loads(x))  # Converting back to Python object

print(deserialized_data.collect())

💡 Issues with RDD Serialization:
✔ Each Python object needs to be serialized before sending over the network.
✔ Deserialization happens every time an action is performed.
✔ This overhead reduces performance significantly when handling large datasets.

🔥 2. Why DataFrames Avoid Serialization Overhead?

• DataFrames use Spark’s Tungsten Execution Engine, which optimizes execution by using off-heap memory and a binary format.
• Instead of serializing entire Python objects, DataFrames store data in an optimized columnar format (Arrow, Parquet, ORC, etc.).
• This avoids the costly process of serializing/deserializing Python objects across the cluster.

Example of DataFrame Avoiding Serialization Overhead

from pyspark.sql import Row

data = [("Alice", 25), ("Bob", 30), ("Charlie", 35)]
df = spark.createDataFrame(data, ["name", "age"])

# Convert DataFrame to RDD (this will now need serialization)
rdd = df.rdd

print(rdd.collect())  # Incurs serialization overhead
print(df.show())  # Avoids serialization overhead

💡 Why DataFrame is Faster?
✔ Uses Tungsten’s binary format (not Python objects).
✔ Uses vectorized execution instead of row-based execution.
✔ Avoids unnecessary serialization by keeping data in an optimized structure.

🔥 3. Performance Benchmark: RDD vs. DataFrame

Let’s test performance with a large dataset.

RDD Serialization Cost

import time

big_rdd = spark.sparkContext.parallelize([(i, i * 2) for i in range(1_000_000)])

start_time = time.time()
rdd_result = big_rdd.map(lambda x: (x[0], x[1] + 10)).collect()
end_time = time.time()

print(f"RDD Execution Time: {end_time - start_time:.3f} seconds")

DataFrame (No Serialization Overhead)

big_df = spark.createDataFrame([(i, i * 2) for i in range(1_000_000)], ["id", "value"])

start_time = time.time()
df_result = big_df.withColumn("value", big_df.value + 10).collect()
end_time = time.time()

print(f"DataFrame Execution Time: {end_time - start_time:.3f} seconds")

📊 Expected Results:

🔥 4. Key Differences Between RDD and DataFrame Serialization

🚀 Key Takeaways

✔ RDDs require serialization of Python objects, leading to performance overhead.
✔ DataFrames use optimized binary formats, avoiding unnecessary serialization.
✔ Tungsten engine + Columnar storage in DataFrames make them much faster.
✔ Use RDDs only when necessary (e.g., for custom transformations, complex aggregations).

🚀 Deep Dive into Tungsten and Catalyst Optimizations in Spark

Apache Spark is optimized for high-performance distributed computing, and two key technologies enable this:
1️⃣ Tungsten Execution Engine – Handles low-level memory management & CPU optimizations.
2️⃣ Catalyst Optimizer – Handles logical and physical query optimizations.

Understanding these optimizations will help you write highly efficient PySpark code. Let’s break it down. 👇

🔥 1. Tungsten Execution Engine (CPU & Memory Optimization)

Before Tungsten, Spark used JVM-based object serialization for data storage & movement, which was slow and memory inefficient.

✅ Tungsten optimizes this using:

• Managed memory allocation (off-heap storage)
• Efficient serialization (binary format, avoiding Java/Python object overhead)
• Whole-stage code generation (WSCG) to optimize CPU execution
• Vectorized processing (processing multiple rows at a time like CPUs do with SIMD instructions)

🛠️ Tungsten Optimization Breakdown

1.1 Off-Heap Memory Management (Avoiding JVM Garbage Collection)

Before Tungsten:

• Spark stored data on JVM heap, causing high GC (Garbage Collection) overhead.
• Frequent object creation & destruction slowed performance.

After Tungsten:

• Stores data off-heap (like Apache Arrow format), avoiding JVM overhead.
• Direct memory access (DMA) reduces serialization time.

💡 Example: RDD vs DataFrame Memory Efficiency

import time
import numpy as np

data = [(i, np.random.rand()) for i in range(1_000_000)]

# RDD - Uses JVM heap memory (Slow)
rdd = spark.sparkContext.parallelize(data)

start = time.time()
rdd_result = rdd.map(lambda x: (x[0], x[1] * 2)).collect()
end = time.time()
print(f"RDD Execution Time: {end - start:.3f} sec")

# DataFrame - Uses Tungsten (Fast)
df = spark.createDataFrame(data, ["id", "value"])

start = time.time()
df_result = df.withColumn("value", df["value"] * 2).collect()
end = time.time()
print(f"DataFrame Execution Time: {end - start:.3f} sec")

✅ Tungsten DataFrame runs ~5x faster than RDD! 🚀

1.2 Whole-Stage Code Generation (WSCG)

• Spark dynamically compiles SQL queries into optimized JVM bytecode.
• This eliminates interpreter overhead and runs optimized machine code.
• Best for complex transformations like aggregations & joins.

💡 Example: RDD vs DataFrame Execution Plan

df.explain(True)  # Shows Tungsten Optimized Plan

🚀 Output:

WholeStageCodegen
+- HashAggregate
   +- Exchange
      +- HashAggregate
         +- Project
            +- Filter
               +- Scan parquet

✅ Spark combines multiple stages into one highly optimized execution block.

1.3 Vectorized Processing (SIMD Optimization)

• Instead of processing rows one by one, Spark processes multiple rows at once (batch processing) using SIMD (Single Instruction Multiple Data).
• This works like GPUs that process multiple pixels at a time.
• Spark’s vectorized Parquet & ORC readers improve performance by ~10x for I/O-heavy workloads.

💡 Example: Enabling Vectorized Processing for Parquet

spark.conf.set("spark.sql.parquet.enableVectorizedReader", "true")

✅ DataFrame operations on Parquet are now 10x faster!

2. Catalyst Optimizer (Logical & Physical Query Optimization)

The Catalyst optimizer is Spark’s query planner that rewrites queries for maximum efficiency. It performs 4 major optimizations:

2.1 Logical Plan Optimization (SQL-like Query Rewriting)

✔ Pushdown Filters: Moves WHERE conditions before scanning data.
✔ Column Pruning: Removes unnecessary columns before loading data.

💡 Example: Query Optimization

df = spark.read.parquet("data.parquet")

# Without optimization: Reads entire dataset
df.select("name").show()

# With optimization: Reads only required columns
df.select("name").explain(True)

🚀 Spark will prune unnecessary columns, reducing I/O.

2.2 Physical Plan Optimization (Execution Strategies)

✔ Broadcast Joins for small datasets (avoiding expensive shuffle joins).
✔ Predicate Pushdown (Applying filters before data is read).

💡 Example: Enabling Broadcast Joins for Faster Execution

from pyspark.sql.functions import broadcast

small_df = spark.read.parquet("small_data.parquet")
large_df = spark.read.parquet("large_data.parquet")

# Normal Join (Slow Shuffle)
result = large_df.join(small_df, "id")

# Broadcast Join (Fast)
result = large_df.join(broadcast(small_df), "id")

✅ Broadcast join avoids expensive shuffling! 🚀

2.3 Adaptive Query Execution (AQE)

Spark dynamically adjusts query execution at runtime for better performance.
✔ Re-optimizes joins dynamically based on data size.
✔ Dynamically coalesces partitions to avoid small file problems.

💡 Enable AQE for Auto-Optimized Queries

spark.conf.set("spark.sql.adaptive.enabled", "true")

✅ Spark will now dynamically optimize partitions & joins!

🔥 3. Comparing RDD vs. DataFrame Performance

🚀 Key Takeaways

✔ Tungsten makes Spark faster by optimizing memory & CPU execution.
✔ Catalyst makes queries smarter by rewriting & optimizing execution plans.
✔ Vectorized execution & SIMD processing give ~10x speedup.
✔ RDDs should be avoided unless necessary—DataFrames & Datasets are much more optimized.
✔ Enabling AQE & Broadcast Joins can drastically improve performance.

import time
from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast

# Initialize Spark Session
spark = SparkSession.builder \
    .appName("Performance Benchmarking") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

# Large dataset for benchmarking
data = [(i, i * 2) for i in range(1_000_000)]

# ------------------- RDD Performance ------------------- #
print("\nBenchmarking RDD Performance...")
rdd = spark.sparkContext.parallelize(data)

# Test RDD Execution Time
start_time = time.time()
rdd_result = rdd.map(lambda x: (x[0], x[1] + 10)).collect()
end_time = time.time()
print(f"RDD Execution Time: {end_time - start_time:.3f} sec")

# ------------------- DataFrame Performance ------------------- #
print("\nBenchmarking DataFrame Performance...")
df = spark.createDataFrame(data, ["id", "value"])

# Test DataFrame Execution Time
start_time = time.time()
df_result = df.withColumn("value", df.value + 10).collect()
end_time = time.time()
print(f"DataFrame Execution Time: {end_time - start_time:.3f} sec")

# ------------------- Broadcast Join vs. Shuffle Join ------------------- #
print("\nBenchmarking Join Performance...")
small_df = spark.createDataFrame([(i, i * 3) for i in range(1000)], ["id", "value"])
large_df = spark.createDataFrame([(i, i * 5) for i in range(1_000_000)], ["id", "value"])

# Shuffle Join (Slow)
start_time = time.time()
shuffle_result = large_df.join(small_df, "id").collect()
end_time = time.time()
print(f"Shuffle Join Execution Time: {end_time - start_time:.3f} sec")

# Broadcast Join (Fast)
start_time = time.time()
broadcast_result = large_df.join(broadcast(small_df), "id").collect()
end_time = time.time()
print(f"Broadcast Join Execution Time: {end_time - start_time:.3f} sec")

# Stop Spark Session
spark.stop()

Pages: 1 2 3 4 5 6 7