manutej/luxor-claude-marketplace/apache-spark-data-processing

Apache Spark Data Processing

A comprehensive skill for mastering Apache Spark data processing, from basic RDD operations to advanced streaming, SQL, and machine learning workflows. Learn to build scalable, distributed data pipelines and analytics systems.

When to Use This Skill

Use Apache Spark when you need to:

• Process Large-Scale Data: Handle datasets too large for single-machine processing (TB to PB scale)
• Perform Distributed Computing: Execute parallel computations across cluster nodes
• Real-Time Stream Processing: Process continuous data streams with low latency
• Complex Data Analytics: Run sophisticated analytics, aggregations, and transformations
• Machine Learning at Scale: Train ML models on massive datasets
• ETL/ELT Pipelines: Build robust data transformation and loading workflows
• Interactive Data Analysis: Perform exploratory analysis on large datasets
• Unified Data Processing: Combine batch and streaming workloads in one framework

Not Ideal For:

• Small datasets (<100 GB) that fit in memory on a single machine
• Simple CRUD operations (use traditional databases)
• Ultra-low latency requirements (<10ms) where specialized stream processors excel
• Workflows requiring strong ACID transactions across distributed data

Core Concepts

Resilient Distributed Datasets (RDDs)

RDDs are Spark's fundamental data abstraction - immutable, distributed collections of objects that can be processed in parallel.

Key Characteristics:

• Resilient: Fault-tolerant through lineage tracking
• Distributed: Partitioned across cluster nodes
• Immutable: Transformations create new RDDs, not modify existing ones
• Lazy Evaluation: Transformations build computation graph; actions trigger execution
• In-Memory Computing: Cache intermediate results for iterative algorithms

RDD Operations:

• Transformations: Lazy operations that return new RDDs (map, filter, flatMap, reduceByKey)
• Actions: Operations that trigger computation and return values (collect, count, reduce, saveAsTextFile)

When to Use RDDs:

• Low-level control over data distribution and partitioning
• Custom partitioning schemes required
• Working with unstructured data (text files, binary data)
• Migrating legacy code from early Spark versions

Prefer DataFrames/Datasets when possible - they provide automatic optimization via Catalyst optimizer.

DataFrames and Datasets

DataFrames are distributed collections of data organized into named columns - similar to a database table or pandas DataFrame, but with powerful optimizations.

DataFrames:

• Structured data with schema
• Automatic query optimization (Catalyst)
• Cross-language support (Python, Scala, Java, R)
• Rich API for SQL-like operations

Datasets (Scala/Java only):

• Typed DataFrames with compile-time type safety
• Best performance in Scala due to JVM optimization
• Combine RDD type safety with DataFrame optimizations

Key Advantages Over RDDs:

• Query Optimization: Catalyst optimizer rewrites queries for efficiency
• Tungsten Execution: Optimized CPU and memory usage
• Columnar Storage: Efficient data representation
• Code Generation: Compile-time bytecode generation for faster execution

Lazy Evaluation

Spark uses lazy evaluation to optimize execution:

1. Transformations build a Directed Acyclic Graph (DAG) of operations
2. Actions trigger execution of the DAG
3. Spark's optimizer analyzes the entire DAG and creates an optimized execution plan
4. Work is distributed across cluster nodes

Benefits:

• Minimize data movement across network
• Combine multiple operations into single stage
• Eliminate unnecessary computations
• Optimize memory usage

Partitioning

Data is divided into partitions for parallel processing:

• Default Partitioning: Typically based on HDFS block size or input source
• Hash Partitioning: Distribute data by key hash (used by groupByKey, reduceByKey)
• Range Partitioning: Distribute data by key ranges (useful for sorted data)
• Custom Partitioning: Define your own partitioning logic

Partition Count Considerations:

• Too few partitions: Underutilized cluster, large task execution time
• Too many partitions: Scheduling overhead, small task execution time
• General rule: 2-4 partitions per CPU core in cluster
• Use repartition() or coalesce() to adjust partition count

Caching and Persistence

Cache frequently accessed data in memory for performance:

# Cache DataFrame in memory
df.cache()  # Shorthand for persist(StorageLevel.MEMORY_AND_DISK)
# Different storage levels
df.persist(StorageLevel.MEMORY_ONLY)       # Fast but may lose data if evicted
df.persist(StorageLevel.MEMORY_AND_DISK)   # Spill to disk if memory full
df.persist(StorageLevel.DISK_ONLY)         # Store only on disk
df.persist(StorageLevel.MEMORY_ONLY_SER)   # Serialized in memory (more compact)
# Unpersist when done
df.unpersist()

When to Cache:

• Data used multiple times in workflow
• Iterative algorithms (ML training)
• Interactive analysis sessions
• Expensive transformations reused downstream

When Not to Cache:

• Data used only once
• Very large datasets that exceed cluster memory
• Streaming applications with continuous new data

Spark SQL

Spark SQL allows you to query structured data using SQL or DataFrame API:

• Execute SQL queries on DataFrames and tables
• Register DataFrames as temporary views
• Join structured and semi-structured data
• Connect to Hive metastore for table metadata
• Support for various data sources (Parquet, ORC, JSON, CSV, JDBC)

Performance Features:

• Catalyst Optimizer: Rule-based and cost-based query optimization
• Tungsten Execution Engine: Whole-stage code generation, vectorized processing
• Adaptive Query Execution (AQE): Runtime optimization based on statistics
• Dynamic Partition Pruning: Skip irrelevant partitions during execution

Broadcast Variables and Accumulators

Shared variables for efficient distributed computing:

Broadcast Variables:

• Read-only variables cached on each node
• Efficient for sharing large read-only data (lookup tables, ML models)
• Avoid sending large data with every task

# Broadcast a lookup table
lookup_table = {"key1": "value1", "key2": "value2"}
broadcast_lookup = sc.broadcast(lookup_table)
# Use in transformations
rdd.map(lambda x: broadcast_lookup.value.get(x, "default"))

Accumulators:

• Write-only variables for aggregating values across tasks
• Used for counters and sums in distributed operations
• Only driver can read final accumulated value

# Create accumulator
error_count = sc.accumulator(0)
# Increment in tasks
rdd.foreach(lambda x: error_count.add(1) if is_error(x) else None)
# Read final value in driver
print(f"Total errors: {error_count.value}")

Spark SQL Deep Dive

DataFrame Creation

Create DataFrames from various sources:

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("SparkSQLExample").getOrCreate()
# From structured data
data = [("Alice", 1), ("Bob", 2), ("Charlie", 3)]
columns = ["name", "id"]
df = spark.createDataFrame(data, columns)
# From files
df_json = spark.read.json("path/to/file.json")
df_parquet = spark.read.parquet("path/to/file.parquet")
df_csv = spark.read.option("header", "true").csv("path/to/file.csv")
# From JDBC sources
df_jdbc = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://host:port/database") \
    .option("dbtable", "table_name") \
    .option("user", "username") \
    .option("password", "password") \
    .load()

DataFrame Operations

Common DataFrame transformations:

# Select columns
df.select("name", "age").show()
# Filter rows
df.filter(df.age > 21).show()
df.where(df["age"] > 21).show()  # Alternative syntax
# Add/modify columns
from pyspark.sql.functions import col, lit
df.withColumn("age_plus_10", col("age") + 10).show()
df.withColumn("country", lit("USA")).show()
# Aggregations
df.groupBy("department").count().show()
df.groupBy("department").agg({"salary": "avg", "age": "max"}).show()
# Sorting
df.orderBy("age").show()
df.orderBy(col("age").desc()).show()
# Joins
df1.join(df2, df1.id == df2.user_id, "inner").show()
df1.join(df2, "id", "left_outer").show()
# Unions
df1.union(df2).show()

SQL Queries

Execute SQL on DataFrames:

# Register DataFrame as temporary view
df.createOrReplaceTempView("people")
# Run SQL queries
sql_result = spark.sql("SELECT name FROM people WHERE age > 21")
sql_result.show()
# Complex queries
result = spark.sql("""
    SELECT
        department,
        COUNT(*) as employee_count,
        AVG(salary) as avg_salary,
        MAX(age) as max_age
    FROM people
    WHERE age > 25
    GROUP BY department
    HAVING COUNT(*) > 5
    ORDER BY avg_salary DESC
""")
result.show()

Data Sources

Spark SQL supports multiple data formats:

Parquet (Recommended for Analytics):

• Columnar storage format
• Excellent compression and query performance
• Schema embedded in file
• Supports predicate pushdown and column pruning

# Write
df.write.parquet("output/path", mode="overwrite", compression="snappy")
# Read with partition pruning
df = spark.read.parquet("output/path").filter(col("date") == "2025-01-01")

ORC (Optimized Row Columnar):

• Similar to Parquet with slightly better compression
• Preferred for Hive integration
• Built-in indexes for faster queries

df.write.orc("output/path", mode="overwrite")
df = spark.read.orc("output/path")

JSON (Semi-Structured Data):

• Human-readable but less efficient
• Schema inference on read
• Good for nested/complex data

# Read with explicit schema
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
schema = StructType([
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True)
])
df = spark.read.schema(schema).json("data.json")

CSV (Legacy/Simple Data):

• Widely compatible but slow
• Requires header inference or explicit schema
• Minimal compression benefits

df.write.csv("output.csv", header=True, mode="overwrite")
df = spark.read.option("header", "true").option("inferSchema", "true").csv("data.csv")

Window Functions

Advanced analytics with window functions:

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, rank, dense_rank, lag, lead, sum, avg
# Define window specification
window_spec = Window.partitionBy("department").orderBy(col("salary").desc())
# Ranking functions
df.withColumn("rank", rank().over(window_spec)).show()
df.withColumn("row_num", row_number().over(window_spec)).show()
df.withColumn("dense_rank", dense_rank().over(window_spec)).show()
# Aggregate functions over window
df.withColumn("dept_avg_salary", avg("salary").over(window_spec)).show()
df.withColumn("running_total", sum("salary").over(window_spec.rowsBetween(Window.unboundedPreceding, Window.currentRow))).show()
# Offset functions
df.withColumn("prev_salary", lag("salary", 1).over(window_spec)).show()
df.withColumn("next_salary", lead("salary", 1).over(window_spec)).show()

User-Defined Functions (UDFs)

Create custom transformations:

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, IntegerType
# Python UDF (slower due to serialization overhead)
def categorize_age(age):
    if age < 18:
        return "Minor"
    elif age < 65:
        return "Adult"
    else:
        return "Senior"
categorize_udf = udf(categorize_age, StringType())
df.withColumn("age_category", categorize_udf(col("age"))).show()
# Pandas UDF (vectorized, faster for large datasets)
from pyspark.sql.functions import pandas_udf
import pandas as pd