🧠 Spark Fundamentals
🎯 What is Apache Spark?
Apache Spark is a unified analytics engine for large-scale data processing. It's 100x faster than MapReduce (in-memory), supports batch, streaming, ML, and graph processing, and provides APIs in Python, Scala, Java, and R. It's the de facto standard for Big Data processing.
Why Spark Matters
Speed: In-memory, 100x faster. Unified: Batch + stream + ML. Simple: High-level APIs. Scalable: Petabytes. Fault-tolerant: RDD lineage. Ecosystem: Spark SQL, MLlib, GraphX.
📌 In-Memory📌 Unified📌 Fast📌 Scalable
# Apache Spark — the big picture
# Spark is a unified analytics engine for Big Data.
# CONCEPTUAL OVERVIEW
print("""
=== Apache Spark Overview ===
Spark is:
- Fast: In-memory processing (100x MapReduce)
- Unified: Batch + Streaming + ML + Graph
- Easy: High-level APIs (Python, Scala, Java, R)
- Scalable: Petabytes of data
- Fault-tolerant: RDD lineage
Core components:
- Spark Core: RDDs, scheduling, memory
- Spark SQL: Structured data, SQL
- Spark Streaming: Real-time data
- MLlib: Machine learning
- GraphX: Graph processing
Key concepts:
- RDD: Resilient Distributed Dataset
- DataFrame: Structured data
- Dataset: Typed DataFrame
- Transformation: Lazy operation
- Action: Triggers computation
- DAG: Directed Acyclic Graph
""")
# WORD COUNT IN SPARK (Python)
print("=== Word Count in Spark (PySpark) ===")
print("""
from pyspark.sql import SparkSession
# Create Spark session
spark = SparkSession.builder \\
.appName("WordCount") \\
.getOrCreate()
# Read text file
lines = spark.read.text("hdfs://data/input.txt")
# Word count
from pyspark.sql.functions import explode, split, col
word_counts = lines \\
.select(explode(split(col("value"), " ")).alias("word")) \\
.groupBy("word") \\
.count() \\
.orderBy("count", ascending=False)
word_counts.show()
# Stop Spark
spark.stop()
""")
print("\n=== Word Count in Spark (RDD API) ===")
print("""
from pyspark import SparkContext
sc = SparkContext("local", "WordCount")
# Read text file
lines = sc.textFile("hdfs://data/input.txt")
# Word count
counts = lines \\
.flatMap(lambda line: line.split()) \\
.map(lambda word: (word.lower(), 1)) \\
.reduceByKey(lambda a, b: a + b)
# Collect results
for word, count in counts.collect():
print(f"{word}: {count}")
sc.stop()
""")
# KEY PROPERTIES
print("""
=== Key Properties of Spark ===
1. IN-MEMORY COMPUTING
- Data cached in RAM
- 100x faster than MapReduce
- For iterative algorithms
2. LAZY EVALUATION
- Transformations are lazy
- Actions trigger computation
- Optimizes execution plan
3. FAULT TOLERANCE
- RDD lineage (recompute lost partitions)
- Checkpointing
- Task retry
4. UNIFIED API
- Batch + Streaming + ML + Graph
- Same code for all workloads
- Python, Scala, Java, R
5. SCALABLE
- Petabytes of data
- Thousands of nodes
- Linear scale
6. DATA LOCALITY
- Move computation to data
- HDFS, S3, Cassandra, etc.
7. RICH ECOSYSTEM
- Spark SQL
- MLlib
- GraphX
- Structured Streaming
""")
# SPARK VS MAPREDUCE
print("=== Spark vs MapReduce ===")
print(f"{'Aspect':<25} {'MapReduce':<25} {'Spark':<25}")
print("-" * 75)
print(f"{'Storage':<25} {'Disk':<25} {'Memory':<25}")
print(f"{'Speed':<25} {'Slow':<25} {'100x faster':<25}")
print(f"{'Iterative':<25} {'Slow':<25} {'Fast':<25}")
print(f"{'API':<25} {'Low-level (Java)':<25} {'High-level (Python, Scala)':<25}")
print(f"{'Processing':<25} {'Batch':<25} {'Batch + Stream + ML':<25}")
print(f"{'Fault tolerance':<25} {'Task retry':<25} {'RDD lineage':<25}")
print(f"{'Memory':<25} {'Low':<25} {'High':<25}")
print(f"{'Cost':<25} {'Lower':<25} {'Higher':<25}")
# REAL-WORLD SCALE
print("\n=== Real-World Spark Deployments ===")
deployments = [
('Netflix', 'Recommendations', 'Petabytes'),
('Uber', 'Real-time analytics', 'Petabytes'),
('Airbnb', 'Search ranking', 'Petabytes'),
('Alibaba', 'Singles Day', 'Petabytes'),
('Facebook', 'ML pipelines', 'Petabytes'),
('Apple', 'Siri, analytics', 'Petabytes'),
]
print(f"{'Company':<15} {'Use Case':<25} {'Scale':<20}")
print("-" * 60)
for company, use_case, scale in deployments:
print(f"{company:<15} {use_case:<25} {scale:<20}")🏗️ Architecture
🎯 Master-Slave Architecture
Spark follows a master-slave architecture: a Driver (master) coordinates work, and Executors (slaves) run tasks. The Cluster Manager (YARN, Mesos, Kubernetes) allocates resources.
Architecture Overview
Driver: Main program, DAG. Executor: Runs tasks. Cluster Manager: Resources. Worker: Physical node. Task: Unit of work. Stage: Task group.
🏗️ Driver🏗️ Executors🏗️ Cluster Manager🏗️ DAG
# Apache Spark Architecture
print("=== Spark Architecture Diagram ===")
print("""
┌─────────────────────────────────────────┐
│ Driver │
│ (SparkSession) │
│ │
│ - Main program │
│ - Creates SparkContext │
│ - Builds DAG │
│ - Schedules tasks │
│ - Coordinates executors │
└─────────────────┬───────────────────────┘
│
│ 1. Request resources
▼
┌─────────────────────────────────────────┐
│ Cluster Manager │
│ (YARN, Mesos, Kubernetes) │
│ │
│ - Allocates resources │
│ - Manages workers │
└─────────────────┬───────────────────────┘
│
┌───────────┼───────────┐
│ │ │
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Worker │ │ Worker │ │ Worker │
│ Node │ │ Node │ │ Node │
│ │ │ │ │ │
│┌───────┐│ │┌───────┐│ │┌───────┐│
││Executor││ ││Executor││ ││Executor││
││(JVM) ││ ││(JVM) ││ ││(JVM) ││
││ ││ ││ ││ ││ ││
││Task 1 ││ ││Task 3 ││ ││Task 5 ││
││Task 2 ││ ││Task 4 ││ ││Task 6 ││
│└───────┘│ │└───────┘│ │└───────┘│
└─────────┘ └─────────┘ └─────────┘
""")
# COMPONENTS
print("\n=== Spark Components ===")
print("""
1. DRIVER
- Main program
- Creates SparkContext/SparkSession
- Defines RDDs/DataFrames
- Builds DAG
- Schedules tasks
- Collects results
2. EXECUTOR
- JVM process on worker node
- Runs tasks
- Stores data (cache)
- Reports to driver
- Multiple per node
3. CLUSTER MANAGER
- Allocates resources
- Options:
- Standalone
- YARN
- Mesos
- Kubernetes
4. WORKER NODE
- Physical machine
- Runs executors
- Has CPU, memory, disk
5. TASK
- Smallest unit of work
- Runs on one partition
- One thread per task
6. STAGE
- Group of tasks
- Separated by shuffle
- Part of a job
7. JOB
- Triggered by action
- Composed of stages
""")
# EXECUTION FLOW
print("\n=== Execution Flow ===")
print("""
1. SUBMIT APPLICATION
- Driver starts
- SparkSession created
2. BUILD DAG
- Transformations define DAG
- Lazy evaluation
3. TRIGGER ACTION
- Action called
- DAG submitted
4. DAG SCHEDULER
- Splits DAG into stages
- Stages separated by shuffle
5. TASK SCHEDULER
- Launches tasks
- One task per partition
- Executors run tasks
6. EXECUTE
- Tasks read data
- Apply transformations
- Write output
7. COLLECT
- Results returned to driver
- Or written to storage
""")
# IMPLEMENTATION
print("\n=== Simulating Spark Architecture ===")
import time
from collections import defaultdict
class SparkDriver:
"""Simulate Spark Driver."""
def __init__(self, app_name):
self.app_name = app_name
self.executors = []
self.jobs = []
self.dag = []
def create_executor(self, executor_id, cores, memory):
"""Create an executor."""
executor = Executor(executor_id, cores, memory)
self.executors.append(executor)
print(f"Driver: Created executor {executor_id} ({cores} cores, {memory} MB)")
return executor
def submit_job(self, job_name, transformations, actions):
"""Submit a job."""
print(f"\nDriver: Submitting job '{job_name}'")
# Build DAG
print(f" Building DAG with {len(transformations)} transformations")
self.dag = transformations
# Split into stages
stages = self._split_stages(transformations)
print(f" Split into {len(stages)} stages")
# Schedule tasks
for stage in stages:
print(f" Stage: {stage['name']} ({len(stage['tasks'])} tasks)")
for task in stage['tasks']:
executor = self._find_executor()
if executor:
executor.run_task(task)
# Execute actions
results = []
for action in actions:
result = action()
results.append(result)
self.jobs.append({
'name': job_name,
'stages': len(stages),
'actions': len(actions),
})
return results
def _split_stages(self, transformations):
"""Split DAG into stages (by shuffle)."""
stages = []
current_stage = {'name': 'Stage 0', 'tasks': []}
for i, t in enumerate(transformations):
if 'shuffle' in t.lower() or 'groupBy' in t:
stages.append(current_stage)
current_stage = {'name': f'Stage {len(stages)}', 'tasks': []}
current_stage['tasks'].append(f"task_{i}")
stages.append(current_stage)
return stages
def _find_executor(self):
"""Find executor with available cores."""
for executor in self.executors:
if executor.available_cores > 0:
return executor
return None
def get_status(self):
return {
'app_name': self.app_name,
'executors': len(self.executors),
'total_cores': sum(e.cores for e in self.executors),
'jobs_completed': len(self.jobs),
}
class Executor:
"""Simulate Spark Executor."""
def __init__(self, executor_id, cores, memory):
self.id = executor_id
self.cores = cores
self.available_cores = cores
self.memory = memory
self.tasks_run = 0
self.cache = {}
def run_task(self, task):
"""Run a task."""
if self.available_cores <= 0:
return None
self.available_cores -= 1
self.tasks_run += 1
# Simulate task execution
time.sleep(0.01)
self.available_cores += 1
return f"{task}_done"
def cache_rdd(self, rdd_id, data):
"""Cache an RDD."""
self.cache[rdd_id] = data
print(f" Executor {self.id}: Cached {rdd_id} ({len(data)} records)")
def get_status(self):
return {
'id': self.id,
'cores': self.cores,
'available_cores': self.available_cores,
'memory_mb': self.memory,
'tasks_run': self.tasks_run,
'cached_rdds': len(self.cache),
}
# Simulate cluster
driver = SparkDriver("wordcount")
# Create executors
for i in range(3):
driver.create_executor(f"exec_{i}", cores=4, memory=4096)
# Submit a job
transformations = [
"read_file",
"flatMap_words",
"map_pair",
"reduceByKey", # shuffle
"sortByKey",
]
actions = [
lambda: "result",
]
results = driver.submit_job("wordcount", transformations, actions)
# Status
print("\n=== Cluster Status ===")
status = driver.get_status()
for key, val in status.items():
print(f" {key}: {val}")
print("\n=== Executor Status ===")
for executor in driver.executors:
status = executor.get_status()
print(f" {status['id']}: {status['tasks_run']} tasks, "
f"{status['available_cores']}/{status['cores']} cores available")
# ARCHITECTURE SUMMARY
print("\n=== Architecture Summary ===")
print(f"{'Component':<25} {'Role':<40}")
print("-" * 65)
print(f"{'Driver':<25} {'Main program, DAG, scheduling':<40}")
print(f"{'Executor':<25} {'Run tasks, cache data':<40}")
print(f"{'Cluster Manager':<25} {'Allocate resources':<40}")
print(f"{'Worker Node':<25} {'Physical machine':<40}")
print(f"{'Task':<25} {'Smallest unit of work':<40}")
print(f"{'Stage':<25} {'Group of tasks':<40}")📦 RDDs
🎯 Resilient Distributed Datasets
RDD (Resilient Distributed Dataset) is Spark's fundamental data structure: an immutable, partitioned collection of records that can be processed in parallel. RDDs are fault-tolerant through lineage.
RDD Properties
Resilient: Fault-tolerant. Distributed: Across nodes. Dataset: Collection of records. Immutable: Read-only. Partitioned: Split for parallelism. Lazy: Computed on demand.
📦 Immutable📦 Partitioned📦 Fault-Tolerant📦 Lazy
# RDDs — Resilient Distributed Datasets
print("=== What is an RDD? ===")
print("""
RDD = Resilient Distributed Dataset
Resilient:
- Fault-tolerant
- Recompute lost partitions
- Lineage graph
Distributed:
- Data spread across nodes
- Parallel processing
- Partitioned
Dataset:
- Collection of records
- Immutable
- Typed (in Scala)
Properties:
- Immutable: Cannot be modified
- Lazy: Computed on demand
- Partitioned: Split for parallelism
- Cached: Can persist in memory
""")
# CREATING RDDS
print("\n=== Creating RDDs ===")
print("""
# From collection (Python list)
data = [1, 2, 3, 4, 5]
rdd = sc.parallelize(data)
# From external file
rdd = sc.textFile("hdfs://data/input.txt")
# From another RDD (transformation)
rdd2 = rdd.map(lambda x: x * 2)
# From DataFrame
df = spark.read.csv("data.csv")
rdd = df.rdd
# From range
rdd = sc.range(1, 1000000)
# From key-value pairs
pairs = [('a', 1), ('b', 2), ('c', 3)]
rdd = sc.parallelize(pairs)
""")
# RDD OPERATIONS
print("\n=== RDD Operations ===")
print("""
Transformations (lazy):
- map(func)
- filter(func)
- flatMap(func)
- distinct()
- union(other)
- intersection(other)
- subtract(other)
- cartesian(other)
- groupByKey()
- reduceByKey(func)
- sortByKey()
- join(other)
- cogroup(other)
- pipe(command)
- coalesce(numPartitions)
- repartition(numPartitions)
Actions (trigger computation):
- collect()
- count()
- first()
- take(n)
- takeSample(withReplacement, n)
- takeOrdered(n, key)
- saveAsTextFile(path)
- saveAsSequenceFile(path)
- countByKey()
- countByValue()
- reduce(func)
- fold(func)
- aggregate(func)
- foreach(func)
""")
# IMPLEMENTATION
print("\n=== Simulating RDD ===")
class RDD:
"""Simulate Spark RDD."""
def __init__(self, data, num_partitions=2, parent=None, operation=None):
self.data = data
self.num_partitions = num_partitions
self.parent = parent
self.operation = operation
self.cached = False
def map(self, func):
"""Map transformation."""
return RDD(
data=self.data,
num_partitions=self.num_partitions,
parent=self,
operation=('map', func)
)
def filter(self, func):
"""Filter transformation."""
return RDD(
data=self.data,
num_partitions=self.num_partitions,
parent=self,
operation=('filter', func)
)
def flatMap(self, func):
"""FlatMap transformation."""
return RDD(
data=self.data,
num_partitions=self.num_partitions,
parent=self,
operation=('flatMap', func)
)
def reduceByKey(self, func):
"""ReduceByKey transformation."""
return RDD(
data=self.data,
num_partitions=self.num_partitions,
parent=self,
operation=('reduceByKey', func)
)
def compute(self):
"""Compute the RDD (lazy evaluation)."""
if self.operation is None:
return self.data
# Get parent data
parent_data = self.parent.compute()
op_type, func = self.operation
if op_type == 'map':
return [func(x) for x in parent_data]
elif op_type == 'filter':
return [x for x in parent_data if func(x)]
elif op_type == 'flatMap':
result = []
for x in parent_data:
result.extend(func(x))
return result
elif op_type == 'reduceByKey':
from collections import defaultdict
grouped = defaultdict(list)
for key, value in parent_data:
grouped[key].append(value)
return [(k, func(v)) for k, v in grouped.items()]
return parent_data
def collect(self):
"""Action: return all data."""
return self.compute()
def count(self):
"""Action: count elements."""
return len(self.compute())
def take(self, n):
"""Action: take first n."""
return self.compute()[:n]
def cache(self):
"""Mark for caching."""
self.cached = True
return self
def lineage(self):
"""Get lineage (DAG)."""
lineage = []
current = self
while current is not None:
if current.operation:
lineage.append(current.operation[0])
current = current.parent
return list(reversed(lineage))
def __repr__(self):
return f"RDD(partitions={self.num_partitions}, cached={self.cached})"
# Create RDD
print("=== Creating RDD ===")
data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
rdd = RDD(data, num_partitions=2)
print(f"RDD: {rdd}")
print(f"Data: {rdd.collect()}")
# Transformations
print("\n=== Transformations (lazy) ===")
mapped = rdd.map(lambda x: x * 2)
print(f"Mapped RDD: {mapped}")
print(f"Lineage: {mapped.lineage()}")
filtered = mapped.filter(lambda x: x > 10)
print(f"Filtered RDD: {filtered}")
print(f"Lineage: {filtered.lineage()}")
# Action triggers computation
print("\n=== Action (triggers computation) ===")
result = filtered.collect()
print(f"Result: {result}")
# Word count with RDD
print("\n=== Word Count with RDD ===")
text_data = [
"hello world hello",
"world of spark",
"hello spark world",
]
text_rdd = RDD(text_data, num_partitions=2)
# Word count pipeline
words = text_rdd.flatMap(lambda line: line.split())
pairs = words.map(lambda word: (word.lower(), 1))
counts = pairs.reduceByKey(lambda a, b: a + b)
print(f"Lineage: {counts.lineage()}")
print(f"Result: {counts.collect()}")
# RDD FAULT TOLERANCE
print("\n=== RDD Fault Tolerance ===")
print("""
Lineage:
- RDD remembers how it was built
- If partition lost, recompute
- No data replication needed
Example:
rdd1 = sc.textFile("data.txt")
rdd2 = rdd1.map(lambda x: x * 2)
rdd3 = rdd2.filter(lambda x: x > 10)
If rdd3 partition lost:
- Recompute from rdd2
- Recompute from rdd1
- Read from HDFS
Checkpointing:
- Save RDD to HDFS
- Break lineage
- Faster recovery
rdd.checkpoint()
""")
# RDD SUMMARY
print("\n=== RDD Summary ===")
print(f"{'Property':<25} {'Description':<40}")
print("-" * 65)
print(f"{'Immutable':<25} {'Cannot be modified':<40}")
print(f"{'Lazy':<25} {'Computed on demand':<40}")
print(f"{'Partitioned':<25} {'Split for parallelism':<40}")
print(f"{'Fault-tolerant':<25} {'Lineage for recovery':<40}")
print(f"{'Cached':<25} {'Can persist in memory':<40}")
print(f"{'Typed':<25} {'In Scala/Java':<40}")📊 DataFrames & Datasets
🎯 Structured Data in Spark
DataFrames and Datasets are higher-level APIs built on RDDs. DataFrames are like tables with named columns; Datasets add type safety. They enable Spark SQL and Catalyst optimization.
DataFrame & Dataset
DataFrame: Table with schema. Dataset: Typed DataFrame. Schema: Column names/types. Catalyst: Query optimizer. Tungsten: Execution engine. API: SQL-like operations.
📊 DataFrame📊 Dataset📊 Schema📊 Catalyst
# DataFrames and Datasets
print("=== What is a DataFrame? ===")
print("""
DataFrame:
- Distributed collection of data
- Organized into named columns
- Like a table in a relational database
- Built on RDDs
- Uses Catalyst optimizer
Analogy:
DataFrame ≈ Table in SQL
Row ≈ Record
Column ≈ Field
Benefits:
- Schema: Named columns with types
- Optimization: Catalyst, Tungsten
- SQL: Query with SQL
- Performance: Faster than RDDs
""")
print("\n=== Creating DataFrames ===")
print("""
# From CSV
df = spark.read.csv("data.csv", header=True, inferSchema=True)
# From JSON
df = spark.read.json("data.json")
# From Parquet
df = spark.read.parquet("data.parquet")
# From RDD
from pyspark.sql import Row
rdd = sc.parallelize([(1, "Alice"), (2, "Bob")])
df = rdd.toDF(["id", "name"])
# From list
data = [(1, "Alice"), (2, "Bob")]
df = spark.createDataFrame(data, ["id", "name"])
# From SQL
df = spark.sql("SELECT * FROM table")
""")
print("\n=== DataFrame Operations ===")
print("""
# Select columns
df.select("name", "age").show()
# Filter rows
df.filter(df.age > 30).show()
df.where("age > 30").show()
# Group and aggregate
df.groupBy("city").count().show()
df.groupBy("city").agg({"salary": "avg"}).show()
# Sort
df.orderBy("age").show()
df.orderBy(df.age.desc()).show()
# Join
df1.join(df2, df1.id == df2.id).show()
# Union
df1.union(df2).show()
# Add column
df.withColumn("new_col", df.age * 2).show()
# Rename column
df.withColumnRenamed("old", "new").show()
# Drop column
df.drop("column").show()
# Distinct
df.distinct().show()
# Limit
df.limit(10).show()
""")
# IMPLEMENTATION
print("\n=== Simulating DataFrame ===")
class DataFrame:
"""Simulate Spark DataFrame."""
def __init__(self, data, columns=None):
self.data = data
self.columns = columns or []
self.plan = []
def select(self, *cols):
"""Select columns."""
self.plan.append(f"select({', '.join(cols)})")
new_data = []
for row in self.data:
new_row = {col: row.get(col) for col in cols}
new_data.append(new_row)
return DataFrame(new_data, list(cols))
def filter(self, condition):
"""Filter rows."""
self.plan.append(f"filter({condition})")
# Simple filter for demo
new_data = [row for row in self.data if eval(condition, {}, row)]
return DataFrame(new_data, self.columns)
def groupBy(self, col):
"""Group by column."""
self.plan.append(f"groupBy({col})")
return GroupedDataFrame(self.data, col)
def orderBy(self, col, ascending=True):
"""Order by column."""
self.plan.append(f"orderBy({col})")
new_data = sorted(self.data, key=lambda r: r[col], reverse=not ascending)
return DataFrame(new_data, self.columns)
def withColumn(self, name, expr):
"""Add column."""
self.plan.append(f"withColumn({name})")
new_data = []
for row in self.data:
new_row = row.copy()
new_row[name] = eval(expr, {}, row)
new_data.append(new_row)
return DataFrame(new_data, self.columns + [name])
def show(self, n=10):
"""Show DataFrame."""
print(f"DataFrame[{len(self.data)} rows]")
print(f"Columns: {self.columns}")
print("-" * 60)
for row in self.data[:n]:
print(row)
if len(self.data) > n:
print(f"... and {len(self.data) - n} more rows")
def collect(self):
"""Collect all rows."""
return self.data
def count(self):
"""Count rows."""
return len(self.data)
def explain(self):
"""Show query plan."""
print("Query Plan:")
print(f" Scan: {len(self.data)} rows")
for step in self.plan:
print(f" → {step}")
class GroupedDataFrame:
"""Grouped DataFrame."""
def __init__(self, data, group_col):
self.data = data
self.group_col = group_col
def count(self):
"""Count per group."""
from collections import Counter
counts = Counter(row[self.group_col] for row in self.data)
result = [{self.group_col: k, 'count': v} for k, v in counts.items()]
return DataFrame(result, [self.group_col, 'count'])
def sum(self, col):
"""Sum per group."""
from collections import defaultdict
sums = defaultdict(int)
for row in self.data:
sums[row[self.group_col]] += row.get(col, 0)
result = [{self.group_col: k, f'sum({col})': v} for k, v in sums.items()]
return DataFrame(result, [self.group_col, f'sum({col})'])
def avg(self, col):
"""Average per group."""
from collections import defaultdict
sums = defaultdict(float)
counts = defaultdict(int)
for row in self.data:
sums[row[self.group_col]] += row.get(col, 0)
counts[row[self.group_col]] += 1
result = [{self.group_col: k, f'avg({col})': sums[k] / counts[k]}
for k in sums]
return DataFrame(result, [self.group_col, f'avg({col})'])
# Create DataFrame
print("=== Creating DataFrame ===")
data = [
{'id': 1, 'name': 'Alice', 'age': 30, 'city': 'NYC', 'salary': 50000},
{'id': 2, 'name': 'Bob', 'age': 25, 'city': 'LA', 'salary': 60000},
{'id': 3, 'name': 'Charlie', 'age': 35, 'city': 'NYC', 'salary': 75000},
{'id': 4, 'name': 'Diana', 'age': 28, 'city': 'LA', 'salary': 55000},
{'id': 5, 'name': 'Eve', 'age': 32, 'city': 'Chicago', 'salary': 80000},
]
df = DataFrame(data, ['id', 'name', 'age', 'city', 'salary'])
df.show()
# Operations
print("\n=== Select ===")
df.select('name', 'age').show()
print("\n=== Filter (age > 30) ===")
df.filter("age > 30").show()
print("\n=== GroupBy city ===")
df.groupBy('city').count().show()
print("\n=== GroupBy city, sum salary ===")
df.groupBy('city').sum('salary').show()
print("\n=== GroupBy city, avg salary ===")
df.groupBy('city').avg('salary').show()
print("\n=== OrderBy salary ===")
df.orderBy('salary', ascending=False).show()
print("\n=== WithColumn ===")
df.withColumn('salary_k', 'salary / 1000').show()
print("\n=== Query Plan ===")
df2 = df.filter("age > 30").select('name', 'salary')
df2.explain()
# DATASET (TYPED)
print("\n=== Datasets (Typed) ===")
print("""
Datasets (Scala/Java):
- Type-safe
- Case classes
- Compile-time checking
Example:
case class Person(name: String, age: Int)
val ds = spark.read.json("people.json").as[Person]
ds.filter(_.age > 30).show()
Benefits:
- Type safety
- Performance (Tungsten)
- Functional API
Note:
Python doesn't have Datasets (dynamic typing)
PySpark uses DataFrames only
""")
# DATAFRAME VS RDD
print("\n=== DataFrame vs RDD ===")
print(f"{'Aspect':<25} {'RDD':<25} {'DataFrame':<25}")
print("-" * 75)
print(f"{'Level':<25} {'Low':<25} {'High':<25}")
print(f"{'Schema':<25} {'No':<25} {'Yes':<25}")
print(f"{'Optimization':<25} {'No':<25} {'Catalyst':<25}")
print(f"{'Performance':<25} {'Slower':<25} {'Faster':<25}")
print(f"{'Type safety':<25} {'Yes (Scala)':<25} {'Yes (Dataset)':<25}")
print(f"{'SQL':<25} {'No':<25} {'Yes':<25}")
print(f"{'Ease of use':<25} {'Harder':<25} {'Easier':<25}")
print(f"{'Use case':<25} {'Low-level':<25} {'Analytics':<25}")🔄 Transformations
🎯 Lazy Operations
Transformations are operations that create a new RDD/DataFrame from an existing one. They are lazy — they don't execute until an action is called. Two types: narrow and wide.
Transformation Types
Narrow: No shuffle, one-to-one. Wide: Shuffle, many-to-many. Map: One-to-one. Filter: Subset. FlatMap: One-to-many. ReduceByKey: Aggregate by key.
🔄 Narrow🔄 Wide🔄 Lazy🔄 Shuffle
# Transformations in Spark
print("=== Narrow Transformations ===")
print("""
Narrow: No shuffle, data stays in partition
1. MAP
- Apply function to each element
- One-to-one
rdd.map(lambda x: x * 2)
2. FILTER
- Keep elements matching predicate
rdd.filter(lambda x: x > 10)
3. FLATMAP
- Apply function, flatten result
- One-to-many
rdd.flatMap(lambda x: x.split())
4. MAP PARTITIONS
- Apply function to each partition
rdd.mapPartitions(func)
5. UNION
- Combine two RDDs
rdd1.union(rdd2)
6. SAMPLE
- Random sample
rdd.sample(withReplacement, fraction)
7. DISTINCT
- Remove duplicates (may shuffle)
rdd.distinct()
""")
print("\n=== Wide Transformations ===")
print("""
Wide: Requires shuffle, data moves across partitions
1. GROUP BY KEY
- Group values by key
rdd.groupByKey()
2. REDUCE BY KEY
- Aggregate values by key
rdd.reduceByKey(lambda a, b: a + b)
3. SORT BY KEY
- Sort by key
rdd.sortByKey()
4. JOIN
- Join two RDDs
rdd1.join(rdd2)
5. COGROUP
- Group values from both RDDs
rdd1.cogroup(rdd2)
6. REPARTITION
- Change number of partitions
rdd.repartition(10)
7. DISTINCT
- Remove duplicates
rdd.distinct()
""")
# IMPLEMENTATION
print("\n=== Simulating Transformations ===")
class SparkRDD:
"""Simulate Spark RDD with transformations."""
def __init__(self, data, num_partitions=2, parent=None, operation=None, is_wide=False):
self.data = data
self.num_partitions = num_partitions
self.parent = parent
self.operation = operation
self.is_wide = is_wide
def map(self, func):
return SparkRDD(
data=self.data, num_partitions=self.num_partitions,
parent=self, operation=('map', func), is_wide=False
)
def filter(self, func):
return SparkRDD(
data=self.data, num_partitions=self.num_partitions,
parent=self, operation=('filter', func), is_wide=False
)
def flatMap(self, func):
return SparkRDD(
data=self.data, num_partitions=self.num_partitions,
parent=self, operation=('flatMap', func), is_wide=False
)
def reduceByKey(self, func):
return SparkRDD(
data=self.data, num_partitions=self.num_partitions,
parent=self, operation=('reduceByKey', func), is_wide=True
)
def groupByKey(self):
return SparkRDD(
data=self.data, num_partitions=self.num_partitions,
parent=self, operation=('groupByKey', None), is_wide=True
)
def sortByKey(self):
return SparkRDD(
data=self.data, num_partitions=self.num_partitions,
parent=self, operation=('sortByKey', None), is_wide=True
)
def compute(self):
if self.operation is None:
return self.data
parent_data = self.parent.compute()
op_type, func = self.operation
if op_type == 'map':
return [func(x) for x in parent_data]
elif op_type == 'filter':
return [x for x in parent_data if func(x)]
elif op_type == 'flatMap':
result = []
for x in parent_data:
result.extend(func(x))
return result
elif op_type == 'reduceByKey':
from collections import defaultdict
grouped = defaultdict(list)
for key, value in parent_data:
grouped[key].append(value)
return [(k, func(v)) for k, v in grouped.items()]
elif op_type == 'groupByKey':
from collections import defaultdict
grouped = defaultdict(list)
for key, value in parent_data:
grouped[key].append(value)
return [(k, list(v)) for k, v in grouped.items()]
elif op_type == 'sortByKey':
return sorted(parent_data, key=lambda x: x[0])
return parent_data
def collect(self):
return self.compute()
def lineage(self):
lineage = []
current = self
while current is not None:
if current.operation:
op_name = current.operation[0]
marker = " [WIDE]" if current.is_wide else ""
lineage.append(f"{op_name}{marker}")
current = current.parent
return list(reversed(lineage))
# Create RDD
data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
rdd = SparkRDD(data, num_partitions=2)
print("=== Narrow Transformations ===")
mapped = rdd.map(lambda x: x * 2)
print(f"map(x*2): {mapped.collect()}")
filtered = rdd.filter(lambda x: x % 2 == 0)
print(f"filter(even): {filtered.collect()}")
flat_mapped = rdd.flatMap(lambda x: [x, x * 10])
print(f"flatMap: {flat_mapped.collect()}")
print("\n=== Wide Transformations ===")
pairs = SparkRDD([('a', 1), ('b', 2), ('a', 3), ('b', 4), ('a', 5)])
print(f"Original: {pairs.collect()}")
reduced = pairs.reduceByKey(lambda a, b: a + b)
print(f"reduceByKey(sum): {reduced.collect()}")
print(f"Lineage: {reduced.lineage()}")
grouped = pairs.groupByKey()
print(f"groupByKey: {grouped.collect()}")
sorted_pairs = pairs.sortByKey()
print(f"sortByKey: {sorted_pairs.collect()}")
# SHUFFLE
print("\n=== Shuffle ===")
print("""
Shuffle: Data movement across partitions
When:
- Wide transformations
- Repartition
- Join
- GroupByKey
Cost:
- Network I/O
- Disk I/O
- Serialization
- Often the bottleneck
Optimization:
- Use reduceByKey instead of groupByKey
- Use map-side combine
- Tune partition count
- Compress shuffle data
Example:
groupByKey: Shuffle all data
reduceByKey: Combine locally first, then shuffle
""")
# TRANSFORMATION SUMMARY
print("\n=== Transformation Summary ===")
print(f"{'Transformation':<20} {'Type':<15} {'Description':<30}")
print("-" * 70)
print(f"{'map':<20} {'Narrow':<15} {'One-to-one':<30}")
print(f"{'filter':<20} {'Narrow':<15} {'Subset':<30}")
print(f"{'flatMap':<20} {'Narrow':<15} {'One-to-many':<30}")
print(f"{'union':<20} {'Narrow':<15} {'Combine RDDs':<30}")
print(f"{'distinct':<20} {'Wide':<15} {'Remove duplicates':<30}")
print(f"{'groupByKey':<20} {'Wide':<15} {'Group by key':<30}")
print(f"{'reduceByKey':<20} {'Wide':<15} {'Aggregate by key':<30}")
print(f"{'sortByKey':<20} {'Wide':<15} {'Sort by key':<30}")
print(f"{'join':<20} {'Wide':<15} {'Join RDDs':<30}")
print(f"{'repartition':<20} {'Wide':<15} {'Change partitions':<30}")⚡ Actions
🎯 Triggering Computation
Actions are operations that trigger computation and return results to the driver or write to storage. They force the lazy transformations to execute.
Action Types
Collect: Return all data. Count: Number of elements. Take: First n. Reduce: Aggregate. Save: Write to storage. Foreach: Side effects.
⚡ Collect⚡ Count⚡ Take⚡ Save
# Actions in Spark
print("=== What is an Action? ===")
print("""
Action:
- Triggers computation
- Returns result to driver
- Or writes to storage
- Forces lazy transformations to execute
Without action:
- Transformations are lazy
- Nothing happens
- Just builds DAG
With action:
- DAG is executed
- Tasks are scheduled
- Results returned
""")
print("\n=== Common Actions ===")
print("""
1. COLLECT
- Return all elements
- Caution: Fits in driver memory?
result = rdd.collect()
2. COUNT
- Number of elements
count = rdd.count()
3. FIRST
- First element
first = rdd.first()
4. TAKE
- First n elements
top10 = rdd.take(10)
5. TAKE SAMPLE
- Random sample
sample = rdd.takeSample(False, 10)
6. TAKE ORDERED
- Top n by ordering
top10 = rdd.takeOrdered(10, key=lambda x: -x)
7. REDUCE
- Aggregate all elements
total = rdd.reduce(lambda a, b: a + b)
8. FOLD
- Like reduce with initial value
total = rdd.fold(0, lambda a, b: a + b)
9. AGGREGATE
- Different types
result = rdd.aggregate((0, 0), seqOp, combOp)
10. FOREACH
- Apply function to each element
- Side effects (no return)
rdd.foreach(print)
11. SAVE AS TEXT FILE
- Write to HDFS
rdd.saveAsTextFile("hdfs://output")
12. COUNT BY KEY
- Count per key
counts = rdd.countByKey()
13. COUNT BY VALUE
- Count per value
counts = rdd.countByValue()
""")
# IMPLEMENTATION
print("\n=== Simulating Actions ===")
class ActionRDD:
"""Simulate Spark RDD with actions."""
def __init__(self, data, parent=None, operation=None):
self.data = data
self.parent = parent
self.operation = operation
self._computed = False
self._result = None
def map(self, func):
return ActionRDD(self.data, parent=self, operation=('map', func))
def filter(self, func):
return ActionRDD(self.data, parent=self, operation=('filter', func))
def compute(self):
if self._computed:
return self._result
if self.operation is None:
self._result = self.data
else:
parent_data = self.parent.compute()
op_type, func = self.operation
if op_type == 'map':
self._result = [func(x) for x in parent_data]
elif op_type == 'filter':
self._result = [x for x in parent_data if func(x)]
self._computed = True
return self._result
# ACTIONS
def collect(self):
print(" [ACTION] collect() — triggering computation")
return self.compute()
def count(self):
print(" [ACTION] count() — triggering computation")
return len(self.compute())
def first(self):
print(" [ACTION] first() — triggering computation")
return self.compute()[0] if self.compute() else None
def take(self, n):
print(f" [ACTION] take({n}) — triggering computation")
return self.compute()[:n]
def reduce(self, func):
print(" [ACTION] reduce() — triggering computation")
data = self.compute()
if not data:
return None
result = data[0]
for x in data[1:]:
result = func(result, x)
return result
def takeOrdered(self, n, key=None):
print(f" [ACTION] takeOrdered({n}) — triggering computation")
data = self.compute()
if key:
return sorted(data, key=key)[:n]
return sorted(data)[:n]
def foreach(self, func):
print(" [ACTION] foreach() — triggering computation")
for x in self.compute():
func(x)
def countByValue(self):
print(" [ACTION] countByValue() — triggering computation")
from collections import Counter
return dict(Counter(self.compute()))
def __repr__(self):
return f"ActionRDD(computed={self._computed})"
# Create RDD
print("=== Creating RDD (lazy) ===")
data = list(range(1, 11))
rdd = ActionRDD(data)
print(f"RDD created: {rdd}")
# Transformations (lazy)
print("\n=== Transformations (lazy) ===")
mapped = rdd.map(lambda x: x * 2)
print(f"After map: {mapped}")
filtered = mapped.filter(lambda x: x > 10)
print(f"After filter: {filtered}")
print("\n=== No computation yet ===")
# Action triggers computation
print("\n=== Actions (trigger computation) ===")
result = filtered.collect()
print(f"Result: {result}")
print(f"RDD after action: {filtered}")
# More actions
print("\n=== More Actions ===")
print(f"count(): {filtered.count()}")
print(f"first(): {filtered.first()}")
print(f"take(3): {filtered.take(3)}")
print(f"reduce(sum): {filtered.reduce(lambda a, b: a + b)}")
print(f"takeOrdered(3, -x): {filtered.takeOrdered(3, key=lambda x: -x)}")
# countByValue
print("\n=== countByValue ===")
words = ActionRDD(['apple', 'banana', 'apple', 'cherry', 'banana', 'apple'])
counts = words.countByValue()
print(f"Counts: {counts}")
# LAZY EVALUATION
print("\n=== Lazy Evaluation Demo ===")
print("""
# This does NOT compute:
rdd = sc.textFile("huge_file.txt")
mapped = rdd.map(lambda x: x * 2)
filtered = mapped.filter(lambda x: x > 10)
# Nothing has happened yet! No computation.
# This DOES compute:
result = filtered.collect()
# Now Spark:
# 1. Builds DAG
# 2. Optimizes
# 3. Schedules tasks
# 4. Executes
# 5. Returns result
Benefits:
- Optimization: Spark can optimize whole DAG
- No wasted work: Only compute what's needed
- Fault tolerance: Recompute from lineage
""")
# ACTION SUMMARY
print("\n=== Action Summary ===")
print(f"{'Action':<25} {'Returns':<25} {'Use Case':<25}")
print("-" * 75)
print(f"{'collect()':<25} {'list':<25} {'All data (small)':<25}")
print(f"{'count()':<25} {'int':<25} {'Count':<25}")
print(f"{'first()':<25} {'element':<25} {'First element':<25}")
print(f"{'take(n)':<25} {'list':<25} {'Sample':<25}")
print(f"{'reduce()':<25} {'element':<25} {'Aggregate':<25}")
print(f"{'takeOrdered()':<25} {'list':<25} {'Top N':<25}")
print(f"{'foreach()':<25} {'None':<25} {'Side effects':<25}")
print(f"{'countByValue()':<25} {'dict':<25} {'Frequency':<25}")
print(f"{'saveAsTextFile()':<25} {'None':<25} {'Write to HDFS':<25}")🦥 Lazy Evaluation & DAG
🎯 Compute Only When Needed
Lazy evaluation means transformations don't execute immediately. Spark builds a DAG (Directed Acyclic Graph) of operations and optimizes it before execution.
DAG & Lazy Eval
Lazy: Transformations don't compute. DAG: Graph of operations. Optimization: Catalyst, Tungsten. Stages: Split by shuffle. Tasks: One per partition. Execution: Triggered by action.
🦥 Lazy📊 DAG⚡ Optimization🔄 Stages
# Lazy Evaluation and DAG
print("=== Lazy Evaluation ===")
print("""
Transformations are lazy:
- Not computed immediately
- Just recorded
- Build DAG
Actions trigger computation:
- DAG is submitted
- Optimized
- Executed
Example:
rdd = sc.textFile("data.txt") # No computation
mapped = rdd.map(lambda x: x * 2) # No computation
filtered = mapped.filter(...) # No computation
result = filtered.collect() # COMPUTE NOW!
""")
print("\n=== DAG (Directed Acyclic Graph) ===")
print("""
DAG:
- Nodes: RDDs
- Edges: Transformations
- Directed: Data flows
- Acyclic: No cycles
Example DAG:
textFile → map → filter → reduceByKey → save
↑ ↑ ↑ ↑
Stage 0 Stage 0 Stage 0 Stage 1 (shuffle)
Stages:
- Split by shuffle boundaries
- Stage 0: textFile → map → filter
- Stage 1: reduceByKey → save
""")
print("\n=== DAG Scheduler ===")
print("""
DAG Scheduler:
1. Receives DAG from action
2. Splits into stages (by shuffle)
3. Creates tasks (one per partition)
4. Submits to Task Scheduler
Stage types:
- ShuffleMapStage: Writes shuffle data
- ResultStage: Computes result
Optimization:
- Pipeline narrow transformations
- Combine into single stage
- Minimize shuffles
""")
# DAG VISUALIZATION
print("\n=== DAG Visualization ===")
print("""
Word Count DAG:
textFile
│
▼
flatMap (split words)
│
▼
map (word, 1)
│
▼
reduceByKey (sum) ← SHUFFLE BOUNDARY
│
▼
saveAsTextFile
Stages:
Stage 0: textFile → flatMap → map
Stage 1: reduceByKey → saveAsTextFile
""")
# IMPLEMENTATION
print("\n=== Simulating DAG ===")
class DAGNode:
"""Node in a DAG."""
def __init__(self, name, operation=None, is_wide=False):
self.name = name
self.operation = operation
self.is_wide = is_wide
self.children = []
self.parents = []
def add_child(self, child):
self.children.append(child)
child.parents.append(self)
class DAGScheduler:
"""Simulate Spark DAG Scheduler."""
def __init__(self):
self.stages = []
def schedule(self, root):
"""Schedule DAG from root."""
print("DAG Scheduler: Analyzing DAG...")
# Find all nodes (topological order)
nodes = []
visited = set()
def traverse(node):
if node.name in visited:
return
visited.add(node.name)
for parent in node.parents:
traverse(parent)
nodes.append(node)
traverse(root)
print(f" Found {len(nodes)} nodes")
# Split into stages
current_stage = []
stage_num = 0
for node in nodes:
if node.is_wide:
if current_stage:
self.stages.append({
'id': stage_num,
'nodes': current_stage,
'type': 'ShuffleMapStage'
})
stage_num += 1
current_stage = []
self.stages.append({
'id': stage_num,
'nodes': [node],
'type': 'ShuffleMapStage'
})
stage_num += 1
else:
current_stage.append(node)
if current_stage:
self.stages.append({
'id': stage_num,
'nodes': current_stage,
'type': 'ResultStage'
})
# Print stages
print(f"\n Split into {len(self.stages)} stages:")
for stage in self.stages:
node_names = [n.name for n in stage['nodes']]
print(f" Stage {stage['id']} ({stage['type']}): {' → '.join(node_names)}")
return self.stages
# Build DAG
print("=== Building DAG ===")
# Create nodes
text_file = DAGNode("textFile")
flat_map = DAGNode("flatMap", is_wide=False)
map_pair = DAGNode("map", is_wide=False)
reduce_by_key = DAGNode("reduceByKey", is_wide=True)
save = DAGNode("saveAsTextFile", is_wide=False)
# Build edges
text_file.add_child(flat_map)
flat_map.add_child(map_pair)
map_pair.add_child(reduce_by_key)
reduce_by_key.add_child(save)
# Schedule
scheduler = DAGScheduler()
stages = scheduler.schedule(save)
# LAZY EVALUATION BENEFITS
print("\n=== Benefits of Lazy Evaluation ===")
print("""
1. OPTIMIZATION
- Whole DAG optimized
- Catalyst optimizer
- Tungsten execution
2. NO WASTED WORK
- Only compute what's needed
- Skip unused branches
3. FAULT TOLERANCE
- Recompute from lineage
- No need to replicate
4. PIPELINING
- Narrow transformations pipelined
- Single pass over data
5. MEMORY EFFICIENCY
- Not materializing intermediates
- Streaming through pipeline
""")
# DAG SUMMARY
print("\n=== DAG Summary ===")
print(f"{'Concept':<25} {'Description':<40}")
print("-" * 65)
print(f"{'DAG':<25} {'Directed Acyclic Graph':<40}")
print(f"{'Lazy':<25} {'Transformations don\\'t compute':<40}")
print(f"{'Action':<25} {'Triggers computation':<40}")
print(f"{'Stage':<25} {'Group of tasks':<40}")
print(f"{'Shuffle':<25} {'Stage boundary':<40}")
print(f"{'Task':<25} {'One per partition':<40}")
print(f"{'Optimization':<25} {'Catalyst, Tungsten':<40}")💾 Caching & Persistence
🎯 Keeping Data in Memory
Caching stores RDDs/DataFrames in memory (or disk) so they can be reused across multiple actions without recomputation. It's key for iterative algorithms.
Caching & Persistence
Cache: In-memory. Persist: Configurable levels. Reuse: Multiple actions. Iterative: ML algorithms. Memory: Limited. Eviction: LRU.
💾 Cache💾 Persist💾 Memory💾 Reuse
# Caching and Persistence
print("=== Why Cache? ===")
print("""
Without caching:
rdd = sc.textFile("data.txt")
mapped = rdd.map(func)
mapped.count() # Computes from scratch
mapped.collect() # Computes AGAIN from scratch
mapped.reduce() # Computes AGAIN from scratch
With caching:
rdd = sc.textFile("data.txt")
mapped = rdd.map(func).cache()
mapped.count() # Computes, caches
mapped.collect() # Uses cache
mapped.reduce() # Uses cache
Benefits:
- Faster (no recomputation)
- Iterative algorithms
- Interactive analysis
""")
print("\n=== Persistence Levels ===")
print("""
StorageLevel options:
1. MEMORY_ONLY
- Store in memory
- If not fit, recompute
- Fastest
- Default for cache()
2. MEMORY_AND_DISK
- Store in memory
- Spill to disk if needed
- Good balance
3. MEMORY_ONLY_SER
- Serialized in memory
- Less space
- More CPU
4. MEMORY_AND_DISK_SER
- Serialized, memory + disk
- Space efficient
5. DISK_ONLY
- Store on disk
- Slowest
- For large data
6. OFF_HEAP
- Off-heap memory
- Avoids GC
7. REPLICATED
- Replicate across nodes
- Fault tolerance
Python:
from pyspark import StorageLevel
rdd.persist(StorageLevel.MEMORY_AND_DISK)
""")
# IMPLEMENTATION
print("\n=== Simulating Caching ===")
class CachedRDD:
"""Simulate RDD with caching."""
def __init__(self, data, parent=None, operation=None):
self.data = data
self.parent = parent
self.operation = operation
self.cached = False
self.cache_data = None
self.compute_count = 0
def map(self, func):
return CachedRDD(self.data, parent=self, operation=('map', func))
def filter(self, func):
return CachedRDD(self.data, parent=self, operation=('filter', func))
def cache(self):
self.cached = True
print(f" Marked for caching")
return self
def compute(self):
"""Compute RDD."""
self.compute_count += 1
if self.cached and self.cache_data is not None:
print(f" [CACHE HIT] Using cached data")
return self.cache_data
print(f" [COMPUTE] Computing (count={self.compute_count})")
if self.operation is None:
result = self.data
else:
parent_data = self.parent.compute()
op_type, func = self.operation
if op_type == 'map':
result = [func(x) for x in parent_data]
elif op_type == 'filter':
result = [x for x in parent_data if func(x)]
if self.cached:
self.cache_data = result
print(f" [CACHED] Stored {len(result)} records")
return result
def collect(self):
return self.compute()
def count(self):
return len(self.compute())
# Without caching
print("=== Without Caching ===")
data = list(range(1, 11))
rdd = CachedRDD(data)
mapped = rdd.map(lambda x: x * 2)
print("\nFirst action:")
r1 = mapped.collect()
print(f"Result: {r1}")
print("\nSecond action:")
r2 = mapped.collect()
print(f"Result: {r2}")
print(f"\nTotal computations: {mapped.compute_count}")
# With caching
print("\n=== With Caching ===")
rdd2 = CachedRDD(data)
mapped2 = rdd2.map(lambda x: x * 2).cache()
print("\nFirst action:")
r1 = mapped2.collect()
print(f"Result: {r1}")
print("\nSecond action:")
r2 = mapped2.collect()
print(f"Result: {r2}")
print("\nThird action:")
r3 = mapped2.collect()
print(f"Result: {r3}")
print(f"\nTotal computations: {mapped2.compute_count}")
print("(Only computed once thanks to caching!)")
# WHEN TO CACHE
print("\n=== When to Cache ===")
print("""
CACHE WHEN:
✓ RDD/DataFrame reused multiple times
✓ Iterative algorithms (ML)
✓ Interactive analysis
✓ Expensive to compute
✓ Fits in memory
DON'T CACHE WHEN:
✗ Used only once
✗ Too large for memory
✗ Cheap to recompute
✗ Data is already cached
Examples:
- K-means: Cache input data
- PageRank: Cache graph
- Interactive: Cache filtered data
- ETL: Don't cache intermediates
""")
# CACHING SUMMARY
print("\n=== Caching Summary ===")
print(f"{'Level':<25} {'Storage':<20} {'Speed':<15}")
print("-" * 60)
print(f"{'MEMORY_ONLY':<25} {'Memory':<20} {'Fastest':<15}")
print(f"{'MEMORY_AND_DISK':<25} {'Memory + Disk':<20} {'Fast':<15}")
print(f"{'MEMORY_ONLY_SER':<25} {'Memory (serialized)':<20} {'Medium':<15}")
print(f"{'DISK_ONLY':<25} {'Disk':<20} {'Slow':<15}")
print(f"{'OFF_HEAP':<25} {'Off-heap':<20} {'Fast':<15}")🔀 Shuffle & Partitioning
🎯 Data Movement Across Partitions
Shuffle is the process of redistributing data across partitions, required by wide transformations. It's often the most expensive operation in Spark.
Shuffle & Partitioning
Shuffle: Data movement. Partition: Data split. Wide: Needs shuffle. Narrow: No shuffle. Cost: Network, disk, CPU. Optimization: Minimize shuffles.
🔀 Shuffle🔀 Partition🔀 Wide🔀 Narrow
# Shuffle and Partitioning
print("=== What is Shuffle? ===")
print("""
Shuffle:
- Redistributes data across partitions
- Required by wide transformations
- Expensive (network, disk, CPU)
When shuffle happens:
- groupByKey
- reduceByKey
- sortByKey
- join
- repartition
- distinct
Why shuffle:
- Need all values for a key together
- Partition by key
- Cannot be done locally
""")
print("\n=== Shuffle Example ===")
print("""
Before shuffle:
Partition 1: (a, 1), (b, 2), (a, 3)
Partition 2: (b, 4), (c, 5), (a, 6)
After shuffle (by key):
Partition 1: (a, 1), (a, 3), (a, 6)
Partition 2: (b, 2), (b, 4)
Partition 3: (c, 5)
Now reduceByKey can work locally on each partition.
""")
print("\n=== Partitioning ===")
print("""
Partitioning:
- Data split into partitions
- One task per partition
- More partitions = more parallelism
Default partitioning:
- HDFS: one partition per block
- Shuffle: 200 partitions (default)
- Parallelize: based on cores
Repartitioning:
rdd.repartition(100) # Shuffle, even distribution
rdd.coalesce(10) # No shuffle, reduce partitions
Partitioner:
- HashPartitioner (default)
- RangePartitioner
- Custom Partitioner
""")
# IMPLEMENTATION
print("\n=== Simulating Shuffle ===")
class ShuffleSimulator:
"""Simulate Spark shuffle."""
def __init__(self, num_partitions=3):
self.num_partitions = num_partitions
self.partitions = [[] for _ in range(num_partitions)]
def add_data(self, partition_id, data):
"""Add data to a partition."""
self.partitions[partition_id].extend(data)
def shuffle(self):
"""Perform shuffle."""
print("\n=== SHUFFLE PHASE ===")
# Collect all data
all_data = []
for i, partition in enumerate(self.partitions):
print(f" Partition {i}: {partition}")
all_data.extend(partition)
# Partition by key hash
new_partitions = [[] for _ in range(self.num_partitions)]
for key, value in all_data:
target = hash(key) % self.num_partitions
new_partitions[target].append((key, value))
print("\n=== AFTER SHUFFLE ===")
for i, partition in enumerate(new_partitions):
print(f" Partition {i}: {partition}")
self.partitions = new_partitions
return new_partitions
def reduce_by_key(self, func):
"""Reduce by key after shuffle."""
print("\n=== REDUCE PHASE ===")
results = []
for i, partition in enumerate(self.partitions):
# Group by key
from collections import defaultdict
grouped = defaultdict(list)
for key, value in partition:
grouped[key].append(value)
# Reduce
for key, values in grouped.items():
result = values[0]
for v in values[1:]:
result = func(result, v)
results.append((key, result))
print(f" Partition {i}: reduce({key}, {values}) = {result}")
return sorted(results)
# Simulate
print("=== Shuffle Simulation ===")
sim = ShuffleSimulator(num_partitions=3)
# Add data (unsorted)
sim.add_data(0, [('a', 1), ('b', 2)])
sim.add_data(1, [('a', 3), ('c', 4)])
sim.add_data(2, [('b', 5), ('a', 6)])
print("Before shuffle:")
for i, p in enumerate(sim.partitions):
print(f" Partition {i}: {p}")
# Shuffle
shuffled = sim.shuffle()
# Reduce
results = sim.reduce_by_key(lambda a, b: a + b)
print(f"\nFinal result: {results}")
# SHUFFLE OPTIMIZATION
print("\n=== Shuffle Optimization ===")
print("""
1. USE REDUCEBYKEY INSTEAD OF GROUPBYKEY
- reduceByKey: combine locally first
- groupByKey: shuffle all data
Bad: groupByKey().mapValues(sum)
Good: reduceByKey(lambda a, b: a + b)
2. INCREASE PARTITIONS
- More partitions = more parallelism
- But too many = overhead
- Rule: 2-4x number of cores
3. USE COMBINER
- Local aggregation before shuffle
- Reduces network traffic
4. COMPRESS SHUFFLE DATA
- Less network traffic
- spark.shuffle.compress=true
5. TUNE SHUFFLE PARAMETERS
- spark.sql.shuffle.partitions
- spark.shuffle.file.buffer
- spark.reducer.maxSizeInFlight
6. AVOID SKEW
- Salting
- Custom partitioner
- Two-phase aggregation
""")
# SHUFFLE SUMMARY
print("\n=== Shuffle Summary ===")
print(f"{'Aspect':<25} {'Description':<40}")
print("-" * 65)
print(f"{'What':<25} {'Redistribute data':<40}")
print(f"{'When':<25} {'Wide transformations':<40}")
print(f"{'Cost':<25} {'Network, disk, CPU':<40}")
print(f"{'Partitions':<25} {'200 (default)':<40}")
print(f"{'Optimization':<25} {'ReduceByKey, combine':<40}")
print(f"{'Tuning':<25} {'Partition count, buffer':<40}")🗄️ Spark SQL
🎯 SQL on Spark
Spark SQL is a Spark module for structured data processing. It provides SQL queries, DataFrames, and Datasets, with the Catalyst optimizer for query optimization.
Spark SQL
SQL: Standard SQL queries. DataFrame: Structured API. Catalyst: Optimizer. Hive: Hive integration. JDBC: Database connectivity. Sources: CSV, JSON, Parquet.
🗄️ SQL🗄️ DataFrame🗄️ Catalyst🗄️ Hive
# Spark SQL
print("=== What is Spark SQL? ===")
print("""
Spark SQL:
- Module for structured data
- SQL queries
- DataFrames and Datasets
- Catalyst optimizer
- Hive integration
Components:
- SparkSession: Entry point
- DataFrame: Structured data
- SQL: Query language
- Catalyst: Optimizer
- Tungsten: Execution engine
""")
print("\n=== Creating SparkSession ===")
print("""
from pyspark.sql import SparkSession
spark = SparkSession.builder \\
.appName("SparkSQL") \\
.config("spark.sql.warehouse.dir", "/user/hive/warehouse") \\
.enableHiveSupport() \\
.getOrCreate()
""")
print("\n=== Reading Data ===")
print("""
# CSV
df = spark.read.csv("data.csv", header=True, inferSchema=True)
# JSON
df = spark.read.json("data.json")
# Parquet
df = spark.read.parquet("data.parquet")
# JDBC
df = spark.read.jdbc(
url="jdbc:postgresql://localhost:5432/db",
table="users",
properties={"user": "user", "password": "pass"}
)
# Hive
df = spark.sql("SELECT * FROM hive_table")
""")
print("\n=== SQL Queries ===")
print("""
# Register as temp view
df.createOrReplaceTempView("users")
# SQL query
result = spark.sql('''
SELECT city, COUNT(*) as count, AVG(salary) as avg_salary
FROM users
WHERE age > 30
GROUP BY city
ORDER BY count DESC
''')
result.show()
""")
print("\n=== DataFrame API ===")
print("""
from pyspark.sql.functions import col, avg, count, desc
result = df \\
.filter(col("age") > 30) \\
.groupBy("city") \\
.agg(
count("*").alias("count"),
avg("salary").alias("avg_salary")
) \\
.orderBy(desc("count"))
result.show()
""")
# IMPLEMENTATION
print("\n=== Simulating Spark SQL ===")
class SparkSQL:
"""Simulate Spark SQL."""
def __init__(self):
self.tables = {}
self.queries = []
def createDataFrame(self, data, name):
"""Create a DataFrame."""
self.tables[name] = data
print(f"Created DataFrame: {name} ({len(data)} rows)")
return DataFrame(data)
def sql(self, query):
"""Execute SQL query."""
self.queries.append(query)
print(f"Executing SQL: {query[:50]}...")
# Simple query parser (for demo)
query_lower = query.lower().strip()
if 'select' in query_lower and 'from' in query_lower:
# Extract table name
from_part = query_lower.split('from')[1].strip().split()[0]
table_name = from_part.strip(';')
if table_name in self.tables:
data = self.tables[table_name]
return self._execute_select(data, query)
return DataFrame([])
def _execute_select(self, data, query):
"""Execute SELECT query."""
query_lower = query.lower()
# Simple WHERE
if 'where' in query_lower:
where_part = query_lower.split('where')[1].split('group')[0].split('order')[0].strip()
# Parse simple condition: column op value
import re
match = re.match(r'(\w+)\s*([<>=!]+)\s*(\d+)', where_part)
if match:
col, op, val = match.groups()
val = int(val)
if op == '>':
data = [r for r in data if r.get(col, 0) > val]
elif op == '<':
data = [r for r in data if r.get(col, 0) < val]
elif op == '=':
data = [r for r in data if r.get(col) == val]
return DataFrame(data)
class DataFrame:
"""Simple DataFrame for Spark SQL simulation."""
def __init__(self, data):
self.data = data
def show(self, n=10):
if not self.data:
print("(empty)")
return
print(f"DataFrame[{len(self.data)} rows]")
for row in self.data[:n]:
print(f" {row}")
if len(self.data) > n:
print(f" ... and {len(self.data) - n} more")
def createOrReplaceTempView(self, name):
print(f"Registered temp view: {name}")
# Demo
print("=== Spark SQL Demo ===")
spark = SparkSQL()
# Create DataFrame
data = [
{'id': 1, 'name': 'Alice', 'age': 30, 'city': 'NYC', 'salary': 50000},
{'id': 2, 'name': 'Bob', 'age': 25, 'city': 'LA', 'salary': 60000},
{'id': 3, 'name': 'Charlie', 'age': 35, 'city': 'NYC', 'salary': 75000},
{'id': 4, 'name': 'Diana', 'age': 28, 'city': 'LA', 'salary': 55000},
{'id': 5, 'name': 'Eve', 'age': 32, 'city': 'Chicago', 'salary': 80000},
]
df = spark.createDataFrame(data, 'users')
df.show()
# SQL query
print("\n=== SQL Query ===")
result = spark.sql("SELECT * FROM users WHERE age > 30")
result.show()
# CATALYST OPTIMIZER
print("\n=== Catalyst Optimizer ===")
print("""
Catalyst:
- Query optimizer
- Rule-based + cost-based
- Analysis → Logical plan → Physical plan
Phases:
1. Analysis: Resolve references
2. Logical optimization: Rule-based
3. Physical planning: Cost-based
4. Code generation: Tungsten
Example:
SELECT * FROM users WHERE age > 30
Logical plan:
Filter(age > 30) → Scan(users)
Optimized:
Push filter down to scan
Column pruning
Predicate pushdown
""")
# SPARK SQL SUMMARY
print("\n=== Spark SQL Summary ===")
print(f"{'Feature':<25} {'Description':<40}")
print("-" * 65)
print(f"{'SQL':<25} {'Standard SQL queries':<40}")
print(f"{'DataFrame':<25} {'Structured API':<40}")
print(f"{'Catalyst':<25} {'Query optimizer':<40}")
print(f"{'Tungsten':<25} {'Execution engine':<40}")
print(f"{'Hive':<25} {'Hive integration':<40}")
print(f"{'JDBC':<25} {'Database connectivity':<40}")🌊 Structured Streaming
🎯 Real-Time Data Processing
Structured Streaming is Spark's stream processing engine built on the DataFrame API. It treats streams as unbounded tables, allowing the same code for batch and streaming.
Structured Streaming
Unbounded: Stream as table. Same API: Batch + stream. Exactly-once: Guaranteed. Event-time: Watermarks. Output: Append, update, complete. Sources: Kafka, files, sockets.
🌊 Streaming🌊 Unbounded🌊 Exactly-once🌊 Watermarks
# Structured Streaming
print("=== What is Structured Streaming? ===")
print("""
Structured Streaming:
- Stream processing engine
- Built on DataFrame API
- Treats stream as unbounded table
- Same code for batch and stream
Key concepts:
- Input table: Stream of data
- Result table: Updated results
- Output modes: Append, Update, Complete
- Triggers: When to process
- Watermarks: Handle late data
""")
print("\n=== Reading Streams ===")
print("""
# From Kafka
df = spark.readStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "localhost:9092") \\
.option("subscribe", "topic") \\
.load()
# From files
df = spark.readStream \\
.format("csv") \\
.option("path", "hdfs://data/stream") \\
.schema(schema) \\
.load()
# From socket
df = spark.readStream \\
.format("socket") \\
.option("host", "localhost") \\
.option("port", 9999) \\
.load()
""")
print("\n=== Stream Processing ===")
print("""
# Word count on stream
from pyspark.sql.functions import explode, split, col
words = df.select(explode(split(col("value"), " ")).alias("word"))
counts = words.groupBy("word").count()
# Output to console
query = counts.writeStream \\
.outputMode("complete") \\
.format("console") \\
.start()
query.awaitTermination()
""")
print("\n=== Output Modes ===")
print("""
1. APPEND
- Only new rows
- Default mode
- For immutable results
2. UPDATE
- Only updated rows
- For aggregations
3. COMPLETE
- All rows
- For aggregations
- Small results
""")
print("\n=== Watermarks ===")
print("""
Watermarks handle late data:
df.withWatermark("timestamp", "10 minutes")
Meaning:
- Wait 10 minutes for late data
- After that, drop late records
- Bounds state
Without watermark:
- State grows forever
- Memory issues
With watermark:
- State bounded
- Memory safe
""")
# IMPLEMENTATION
print("\n=== Simulating Streaming ===")
import time
import threading
from collections import deque, defaultdict
class StreamProcessor:
"""Simulate Structured Streaming."""
def __init__(self, window_size=10):
self.buffer = deque(maxlen=100)
self.window = deque(maxlen=window_size)
self.results = defaultdict(int)
self.running = False
self.processed = 0
def add_event(self, event):
"""Add event to stream."""
self.buffer.append(event)
def process_batch(self):
"""Process a batch of events."""
while self.buffer:
event = self.buffer.popleft()
self.window.append(event)
self.results[event] += 1
self.processed += 1
def get_results(self):
"""Get current results."""
return dict(self.results)
def reset(self):
"""Reset state."""
self.results.clear()
self.processed = 0
# Simulate stream
print("=== Streaming Simulation ===")
processor = StreamProcessor()
# Simulate events
events = ['apple', 'banana', 'apple', 'cherry', 'banana', 'apple',
'cherry', 'date', 'apple', 'banana']
print("Processing events:")
for event in events:
processor.add_event(event)
processor.process_batch()
print(f" Event: {event}, Results: {processor.get_results()}")
print(f"\nFinal results:")
for word, count in sorted(processor.results.items()):
print(f" {word}: {count}")
print(f"Total processed: {processor.processed}")
# STREAMING VS BATCH
print("\n=== Streaming vs Batch ===")
print(f"{'Aspect':<25} {'Batch':<25} {'Streaming':<25}")
print("-" * 75)
print(f"{'Input':<25} {'Bounded':<25} {'Unbounded':<25}")
print(f"{'Latency':<25} {'Minutes-hours':<25} {'Seconds-ms':<25}")
print(f"{'Processing':<25} {'All at once':<25} {'Continuous':<25}")
print(f"{'API':<25} {'DataFrame':<25} {'DataFrame':<25}")
print(f"{'Fault tolerance':<25} {'Task retry':<25} {'Exactly-once':<25}")
print(f"{'Use case':<25} {'Analytics':<25} {'Real-time':<25}")
# STREAMING SUMMARY
print("\n=== Structured Streaming Summary ===")
print(f"{'Concept':<25} {'Description':<40}")
print("-" * 65)
print(f"{'Unbounded table':<25} {'Stream as table':<40}")
print(f"{'Output modes':<25} {'Append, Update, Complete':<40}")
print(f"{'Triggers':<25} {'When to process':<40}")
print(f"{'Watermarks':<25} {'Handle late data':<40}")
print(f"{'Exactly-once':<25} {'Guaranteed processing':<40}")
print(f"{'Sources':<25} {'Kafka, files, sockets':<40}")
print(f"{'Sinks':<25} {'Console, Kafka, files':<40}")🤖 MLlib
🎯 Machine Learning on Spark
MLlib is Spark's machine learning library. It provides scalable algorithms for classification, regression, clustering, and more, designed for distributed data.
MLlib
Algorithms: Classification, regression. Clustering: K-means, LDA. Recommendation: ALS. Pipelines: ML pipelines. Feature: Extraction, transformation. Evaluation: Metrics.
🤖 Classification🤖 Regression🤖 Clustering🤖 Pipelines
# MLlib — Machine Learning on Spark
print("=== MLlib Overview ===")
print("""
MLlib:
- Scalable ML library
- Distributed algorithms
- DataFrame-based API (ml)
- RDD-based API (mllib, legacy)
Algorithms:
- Classification: Logistic Regression, Decision Trees, Random Forest, GBT, SVM, Naive Bayes
- Regression: Linear, Generalized Linear, Decision Trees, Random Forest
- Clustering: K-means, LDA, Bisecting K-means, Gaussian Mixture
- Collaborative Filtering: ALS
- Dimensionality Reduction: PCA, SVD
- Feature: Tokenizer, VectorAssembler, StandardScaler
- Pipelines: Pipeline, CrossValidator
""")
print("\n=== ML Pipeline ===")
print("""
from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.evaluation import BinaryClassificationEvaluator
# Load data
df = spark.read.csv("data.csv", header=True, inferSchema=True)
# Feature engineering
assembler = VectorAssembler(
inputCols=["age", "salary"],
outputCol="features"
)
scaler = StandardScaler(
inputCol="features",
outputCol="scaled_features"
)
# Model
lr = LogisticRegression(
featuresCol="scaled_features",
labelCol="label"
)
# Pipeline
pipeline = Pipeline(stages=[assembler, scaler, lr])
# Train/test split
train, test = df.randomSplit([0.8, 0.2])
# Train
model = pipeline.fit(train)
# Predict
predictions = model.transform(test)
predictions.show()
# Evaluate
evaluator = BinaryClassificationEvaluator(labelCol="label")
auc = evaluator.evaluate(predictions)
print(f"AUC: {auc}")
""")
# IMPLEMENTATION
print("\n=== Simulating MLlib ===")
class MLPipeline:
"""Simulate Spark ML Pipeline."""
def __init__(self, stages=None):
self.stages = stages or []
self.model = None
def fit(self, data):
"""Train the pipeline."""
print("Training pipeline...")
transformed = data
for stage in self.stages[:-1]: # All but last
print(f" Applying {stage['name']}")
transformed = stage['transform'](transformed)
# Train final stage (model)
model_stage = self.stages[-1]
print(f" Training {model_stage['name']}")
self.model = model_stage['fit'](transformed)
return self
def transform(self, data):
"""Make predictions."""
print("Making predictions...")
transformed = data
for stage in self.stages[:-1]:
transformed = stage['transform'](transformed)
return self.model['predict'](transformed)
# Simple ML example: Linear regression
print("=== Simple Linear Regression ===")
# Training data: y = 2x + 1
train_data = [
{'x': 1, 'y': 3},
{'x': 2, 'y': 5},
{'x': 3, 'y': 7},
{'x': 4, 'y': 9},
{'x': 5, 'y': 11},
]
# Feature assembler
def assemble_features(data):
return [{'features': [r['x']], 'label': r['y']} for r in data]
# Linear regression model
def fit_linear(data):
# Simple least squares
n = len(data)
sum_x = sum(r['features'][0] for r in data)
sum_y = sum(r['label'] for r in data)
sum_xy = sum(r['features'][0] * r['label'] for r in data)
sum_x2 = sum(r['features'][0] ** 2 for r in data)
slope = (n * sum_xy - sum_x * sum_y) / (n * sum_x2 - sum_x ** 2)
intercept = (sum_y - slope * sum_x) / n
return {'slope': slope, 'intercept': intercept}
def predict_linear(model, data):
return [{
'features': r['features'],
'prediction': model['slope'] * r['features'][0] + model['intercept']
} for r in data]
# Build pipeline
pipeline = MLPipeline([
{'name': 'VectorAssembler', 'transform': assemble_features},
{'name': 'LinearRegression', 'fit': fit_linear},
])
# Train
model = pipeline.fit(train_data)
print(f"\nModel: y = {model.model['slope']:.2f}x + {model.model['intercept']:.2f}")
# Predict
test_data = [{'x': 6}, {'x': 7}, {'x': 8}]
predictions = model.transform(test_data)
print("\nPredictions:")
for p in predictions:
print(f" x={p['features'][0]}, y={p['prediction']:.2f}")
# ML ALGORITHMS
print("\n=== MLlib Algorithms ===")
print(f"{'Category':<20} {'Algorithms':<50}")
print("-" * 70)
print(f"{'Classification':<20} {'Logistic Regression, Decision Trees':<50}")
print(f"{'':<20} {'Random Forest, GBT, SVM, Naive Bayes':<50}")
print(f"{'Regression':<20} {'Linear, GLM, Decision Trees':<50}")
print(f"{'Clustering':<20} {'K-means, LDA, Bisecting K-means':<50}")
print(f"{'Recommendation':<20} {'ALS':<50}")
print(f"{'Dim. Reduction':<20} {'PCA, SVD':<50}")
print(f"{'Feature':<20} {'Tokenizer, VectorAssembler, Scaler':<50}")
# MLlib SUMMARY
print("\n=== MLlib Summary ===")
print(f"{'Aspect':<25} {'Description':<40}")
print("-" * 65)
print(f"{'API':<25} {'DataFrame-based (ml)':<40}")
print(f"{'Algorithms':<25} {'Classification, regression':<40}")
print(f"{'Pipelines':<25} {'End-to-end ML workflows':<40}")
print(f"{'Evaluation':<25} {'Metrics, cross-validation':<40}")
print(f"{'Scale':<25} {'Distributed training':<40}")🕸️ GraphX
🎯 Graph Processing on Spark
GraphX is Spark's graph processing library. It provides a distributed graph abstraction and algorithms for PageRank, connected components, and more.
GraphX
Graph: Vertices + edges. VertexRDD: Vertices. EdgeRDD: Edges. Triplets: Edge + vertices. Algorithms: PageRank, SSSP. Pregel: Graph processing.
🕸️ Graph🕸️ Vertices🕸️ Edges🕸️ Pregel
# GraphX — Graph Processing
print("=== What is GraphX? ===")
print("""
GraphX:
- Graph processing library
- Distributed graphs
- Built on Spark
- Pregel API
Graph:
- Vertices: Nodes (users, pages)
- Edges: Relationships (follows, links)
- Properties: Data on vertices/edges
Use cases:
- Social networks
- Web graphs
- Recommendation
- Fraud detection
- Knowledge graphs
""")
print("\n=== Creating a Graph ===")
print("""
from pyspark.graphx import Graph, VertexRDD, EdgeRDD
# Vertices: (id, attribute)
vertices = sc.parallelize([
(1, "Alice"),
(2, "Bob"),
(3, "Charlie"),
(4, "Diana")
])
# Edges: (src, dst, attribute)
edges = sc.parallelize([
(1, 2, "follows"),
(2, 3, "follows"),
(3, 1, "follows"),
(3, 4, "follows"),
(4, 1, "follows")
])
# Create graph
graph = Graph(vertices, edges)
# Basic operations
graph.vertices.collect()
graph.edges.collect()
graph.degrees.collect()
graph.inDegrees.collect()
graph.outDegrees.collect()
""")
print("\n=== Graph Algorithms ===")
print("""
# PageRank
pagerank = graph.pageRank(tol=0.01)
pagerank.vertices.collect()
# Connected Components
components = graph.connectedComponents()
components.vertices.collect()
# Triangle Count
triangles = graph.triangleCount()
triangles.vertices.collect()
# Shortest Paths
shortest = graph.shortestPaths(landmarks=[1])
shortest.vertices.collect()
# Label Propagation
labels = graph.labelPropagation(maxIter=5)
labels.vertices.collect()
""")
print("\n=== Pregel API ===")
print("""
Pregel:
- Vertex-centric programming
- Iterative graph processing
- Message passing
Phases:
1. Receive messages
2. Compute new value
3. Send messages
4. Repeat until convergence
Example: Shortest Path
def vertexProgram(id, dist, msg):
return min(dist, msg)
def sendMessage(edge):
if edge.srcAttr + edge.attr < edge.dstAttr:
return (edge.dstId, edge.srcAttr + edge.attr)
else:
return None
graph.pregel(initialMsg=float('inf'),
maxIter=10,
activeDirection="out",
vertexProgram=vertexProgram,
sendMsg=sendMessage)
""")
# IMPLEMENTATION
print("\n=== Simulating GraphX ===")
class Graph:
"""Simple graph implementation."""
def __init__(self, vertices, edges):
self.vertices = vertices # {id: attr}
self.edges = edges # [(src, dst, attr)]
def degrees(self):
"""Compute degrees."""
degrees = defaultdict(int)
for src, dst, _ in self.edges:
degrees[src] += 1
degrees[dst] += 1
return dict(degrees)
def in_degrees(self):
"""Compute in-degrees."""
in_deg = defaultdict(int)
for src, dst, _ in self.edges:
in_deg[dst] += 1
return dict(in_deg)
def out_degrees(self):
"""Compute out-degrees."""
out_deg = defaultdict(int)
for src, dst, _ in self.edges:
out_deg[src] += 1
return dict(out_deg)
def pagerank(self, iterations=10, damping=0.85):
"""Compute PageRank."""
n = len(self.vertices)
ranks = {v: 1.0 / n for v in self.vertices}
for _ in range(iterations):
new_ranks = {}
for v in self.vertices:
# Sum from incoming edges
incoming = sum(ranks[src] / self.out_degrees().get(src, 1)
for src, dst, _ in self.edges if dst == v)
new_ranks[v] = (1 - damping) / n + damping * incoming
ranks = new_ranks
return ranks
def connected_components(self):
"""Find connected components."""
# Union-Find
parent = {v: v for v in self.vertices}
def find(x):
while parent[x] != x:
parent[x] = parent[parent[x]]
x = parent[x]
return x
def union(x, y):
px, py = find(x), find(y)
if px != py:
parent[px] = py
for src, dst, _ in self.edges:
union(src, dst)
components = defaultdict(list)
for v in self.vertices:
components[find(v)].append(v)
return dict(components)
def triangle_count(self):
"""Count triangles."""
# Build adjacency
adj = defaultdict(set)
for src, dst, _ in self.edges:
adj[src].add(dst)
adj[dst].add(src)
count = 0
for src, dst, _ in self.edges:
# Common neighbors
common = adj[src] & adj[dst]
count += len(common)
return count // 3 # Each triangle counted 3 times
# Create graph
print("=== Graph Demo ===")
vertices = {1: "Alice", 2: "Bob", 3: "Charlie", 4: "Diana"}
edges = [
(1, 2, "follows"),
(2, 3, "follows"),
(3, 1, "follows"),
(3, 4, "follows"),
(4, 1, "follows"),
]
graph = Graph(vertices, edges)
print("Vertices:", graph.vertices)
print("Edges:", graph.edges)
print("\nDegrees:", graph.degrees())
print("In-degrees:", graph.in_degrees())
print("Out-degrees:", graph.out_degrees())
# PageRank
print("\nPageRank:")
ranks = graph.pagerank(iterations=10)
for v, rank in sorted(ranks.items(), key=lambda x: -x[1]):
print(f" {vertices[v]}: {rank:.4f}")
# Connected components
print("\nConnected components:")
components = graph.connected_components()
for root, members in components.items():
names = [vertices[v] for v in members]
print(f" Component: {names}")
# Triangle count
print(f"\nTriangle count: {graph.triangle_count()}")
# GRAPHX SUMMARY
print("\n=== GraphX Summary ===")
print(f"{'Feature':<25} {'Description':<40}")
print("-" * 65)
print(f"{'Graph':<25} {'Vertices + edges':<40}")
print(f"{'VertexRDD':<25} {'Vertices with attributes':<40}")
print(f"{'EdgeRDD':<25} {'Edges with attributes':<40}")
print(f"{'Triplets':<25} {'Edge + source + dest':<40}")
print(f"{'Pregel':<25} {'Vertex-centric processing':<40}")
print(f"{'PageRank':<25} {'Importance ranking':<40}")
print(f"{'Connected components':<25} {'Component finding':<40}")🚀 Future: Spark in Modern Computing
🎯 Where Spark Is Going
Spark continues to evolve: cloud-native, serverless, AI/ML integration, and unified analytics. It's the dominant Big Data processing framework and keeps expanding.
Future Directions
Cloud: Managed services. Serverless: No infrastructure. AI: ML pipelines. Unified: Batch + stream. Lakehouse: Delta Lake. Kubernetes: Container-native.
☁️ Cloud⚡ Serverless🤖 AI🏠 Lakehouse
# Future of Apache Spark
print("=== 1. Cloud-Native Spark ===")
print("""
Trend: Spark on cloud
Managed services:
- AWS EMR
- Google Dataproc
- Azure HDInsight
- Databricks
Benefits:
- No infrastructure
- Auto-scaling
- Pay per use
Serverless:
- AWS Glue
- Google Cloud Dataproc Serverless
- Azure Synapse
Future:
- Cloud-native
- Serverless
- Event-driven
""")
print("\n=== 2. AI and ML Integration ===")
print("""
Trend: Spark + AI
Integrations:
- MLlib (built-in)
- TensorFlow on Spark
- Horovod (distributed training)
- PyTorch on Spark
- MLflow (experiment tracking)
Use cases:
- Feature engineering
- Model training
- Batch inference
- MLOps
Future:
- Unified ML platform
- AutoML
- Feature stores
""")
print("\n=== 3. Lakehouse Architecture ===")
print("""
Trend: Delta Lake + Spark
Delta Lake:
- ACID transactions
- Schema evolution
- Time travel
- Unified batch + stream
Spark + Delta:
- Data lake + warehouse
- One copy of data
- BI + ML
Future:
- Open table formats
- Multi-engine access
- Unified governance
""")
print("\n=== 4. Kubernetes-Native Spark ===")
print("""
Trend: Spark on Kubernetes
Benefits:
- Container-native
- Resource isolation
- Cloud-agnostic
Features:
- Dynamic allocation
- Pod templates
- Volume mounts
Future:
- Kubernetes as default
- Spark Operator
- Serverless Spark
""")
print("\n=== 5. Unified Analytics ===")
print("""
Trend: Batch + Stream + ML unified
Spark 3.x:
- Structured Streaming
- Spark SQL
- MLlib
- GraphX
Future:
- Same code for all
- Lambda architecture simplified
- Kappa architecture
Benefits:
- Less code
- Easier maintenance
- Consistent semantics
""")
print("\n=== 6. Performance Improvements ===")
print("""
Trend: Faster Spark
Spark 3.x:
- Adaptive Query Execution (AQE)
- Dynamic partition pruning
- GPU acceleration
- Photon (Databricks)
Future:
- Better optimizer
- GPU-native
- Vectorized execution
- Native code generation
""")
# TIMELINE
print("\n=== Timeline ===")
print("""
Past:
2009: Spark started (Berkeley)
2010: Open sourced
2013: Apache incubator
2014: Spark 1.0
2016: Spark 2.0 (DataFrames)
2020: Spark 3.0 (AQE)
Present:
2024: Spark 3.5
2025: Spark dominant
Future:
2030: Serverless Spark?
2035: AI-native Spark?
2040: New paradigms?
""")
print("\n=== Final Thoughts ===")
print("""
Apache Spark is:
- The dominant Big Data framework
- Fast, unified, scalable
- Evolving with cloud and AI
- Essential for modern data engineering
The future is:
- Cloud-native
- Serverless
- AI-powered
- Unified
- Lakehouse
"Spark is the operating system
for Big Data."
Keep exploring. Keep building.
""")