Apache Spark Core Concepts for DP-900

Apache Spark is an open-source, distributed computing framework designed for fast, in-memory data processing. Unlike Hadoop MapReduce, which writes intermediate results to disk, Spark keeps data in memory, reducing I/O overhead and accelerating iterative algorithms and interactive queries. Spark provides APIs in Scala, Java, Python, and R, and supports batch processing, streaming, machine learning, and graph processing through its ecosystem libraries.

A Spark application runs as a set of independent processes coordinated by a driver program. The driver is the main control process that:

Executors are worker processes that:

Each Spark cluster has one driver and multiple executors. The number of executors, their memory, and cores are configurable via parameters like spark.executor.instances, spark.executor.memory, and spark.executor.cores. For example, a common configuration on Azure Synapse might set:

spark.executor.instances=10
spark.executor.memory=8g
spark.executor.cores=4

This means 10 executors, each with 8 GB RAM and 4 CPU cores, capable of running 4 tasks in parallel each.

RDD is the fundamental data structure of Spark. It is an immutable, partitioned collection of records that can be operated on in parallel. Key characteristics: - Resilient: If a partition is lost due to node failure, it can be recomputed from lineage (the sequence of transformations that created it). - Distributed: Data is split into partitions across multiple nodes. - Dataset: A collection of typed objects (e.g., RDD[String], RDD[(Int, String)]).

RDDs support two types of operations: 1. Transformations: Lazy operations that produce a new RDD (e.g., map, filter, flatMap, reduceByKey). They are not executed until an action is called. 2. Actions: Operations that trigger computation and return a value to the driver or write data to storage (e.g., count, collect, saveAsTextFile).

Example transformation:

rdd = sc.textFile("wasbs://container@storage.blob.core.windows.net/data.txt")
words = rdd.flatMap(lambda line: line.split(" "))
word_pairs = words.map(lambda w: (w, 1))

No computation happens until an action like word_pairs.reduceByKey(lambda a,b: a+b).collect() is called.

DataFrames are a higher-level abstraction built on top of RDDs, providing a schema (column names and types) and allowing SQL-like queries. DataFrames are more efficient than RDDs due to Catalyst optimizer and Tungsten execution engine. In PySpark, you create a DataFrame from various sources:

df = spark.read.format("parquet").load("path/to/data")
df.createOrReplaceTempView("my_table")
spark.sql("SELECT column1, COUNT(*) FROM my_table GROUP BY column1").show()

DataFrames support both transformations (e.g., select, filter, groupBy, join) and actions (e.g., show, count, write).

Spark uses lazy evaluation: transformations are not executed immediately but are recorded as a lineage graph. When an action is called, the driver builds a DAG of stages, where each stage contains a set of tasks that can be executed together (i.e., without shuffling data across nodes). The DAG Scheduler: 1. Identifies the optimal execution plan. 2. Partitions the DAG into stages based on shuffle dependencies. 3. Submits tasks to the Task Scheduler, which launches them on executors.

For example, consider:

rdd = sc.textFile("file.txt")
rdd2 = rdd.filter(lambda line: line.startswith("ERROR"))
rdd3 = rdd2.map(lambda line: line.split("\t")[1])
result = rdd3.count()

The DAG will have two stages: Stage 0 reads the file and filters; Stage 1 maps and counts. The shuffle boundary occurs because count is an action that aggregates across partitions.

A shuffle is a process of redistributing data across partitions, often required by operations like groupByKey, reduceByKey, join, and distinct. Shuffles are expensive because they involve disk I/O, network transfer, and serialization. Spark writes map output to local disk, then fetches it by reducers. The number of shuffle partitions is controlled by spark.sql.shuffle.partitions (default 200).

Partitioning determines how data is split across nodes. Default partitioning is based on block size (e.g., 128 MB for HDFS). You can repartition or coalesce:

df.repartition(100, "key_column")  # increases partitions, full shuffle
df.coalesce(10)  # decreases partitions, no full shuffle

Caching stores intermediate results in memory to speed up iterative algorithms. You can persist an RDD or DataFrame at different storage levels: - MEMORY_ONLY: Deserialize objects in memory. If not enough memory, partitions are recomputed. - MEMORY_AND_DISK: Store in memory; spill to disk if necessary. - DISK_ONLY: Store on disk only. - MEMORY_ONLY_SER (Java/Scala): Serialized objects in memory, less memory but more CPU. - MEMORY_AND_DISK_SER: Like above but spills to disk.

Example:

df.persist(StorageLevel.MEMORY_AND_DISK)
# or
spark.conf.set("spark.sql.adaptive.enabled", "true")  # for automatic caching

Azure Synapse Analytics provides a serverless Spark pool and dedicated Spark pools. You can create a Spark pool with specific node sizes (e.g., Medium: 4 vCores, 32 GB RAM). Serverless pools automatically scale based on workload. Azure Databricks is a managed Spark platform with optimized runtime (e.g., Databricks Runtime 10.4 LTS includes Spark 3.2.1). Both integrate with Azure Data Lake Storage Gen2 and Azure Blob Storage.

Spark provides a web UI on port 4040 (driver) that shows:

You can also access Spark logs via Azure Monitor or Databricks cluster logs.

A large retailer uses Azure Synapse Analytics to process daily sales data from 10,000 stores. Raw data lands in Azure Data Lake Storage Gen2 as CSV files (approx. 5 TB per day). A Spark job runs nightly to:

The pipeline uses 50 executors with 8 GB memory each. Caching the product catalog DataFrame (which is only 2 GB) speeds up the join significantly. Without caching, the catalog is read from disk each time, adding 10 minutes to the 2-hour job. Misconfiguration: setting spark.sql.shuffle.partitions to 50 (too low) caused severe skew and OOM errors; increasing to 500 resolved the issue.

A financial services company uses Azure Databricks to detect fraudulent credit card transactions in near real-time. Transactions arrive as a stream from Azure Event Hubs (thousands per second). The Spark Structured Streaming job:

The job runs with 8 executors, each with 4 cores. The checkpoint location is set to ADLS to allow recovery from failures. Key configuration: spark.sql.streaming.schemaInference set to true for flexible schema. A common pitfall is not setting spark.sql.streaming.checkpointLocation, which causes loss of state on restart.

A healthcare research team uses Azure Synapse Serverless Spark to explore genomic data. They have 20 TB of variant call data in Parquet format. Data scientists run ad-hoc queries using Python notebooks. The serverless pool auto-scales from 5 to 50 executors based on workload. They use DataFrames and Spark SQL extensively. For example:

SELECT gene, COUNT(*) as variant_count
FROM variants
WHERE population = 'EUR'
GROUP BY gene
ORDER BY variant_count DESC

The query completes in 30 seconds thanks to the Catalyst optimizer and columnar storage. However, a common mistake is using collect() on large results, which overwhelms the driver memory. Instead, they use show(10) or write to a table.