Home › Enterprise Intelligence Architecture

Data Fabric & Distributed Compute

The MapReduce and Spark execution model, partitioning and data skew, the real cost of shuffle, caching and load balancing — the engineering physics beneath every enterprise data platform.

Last reviewed: 2026-09-04 · Українською

In this area

Layer 1 of an enterprise data platform is not a framework choice — it is physics. The network costs more than disk, disk costs more than memory, and a stage finishes at the speed of its slowest task, never its average one. This module works through the three levers an architect actually controls: how many bytes are read, how many of them cross the network, and how evenly they are spread. Everything above — the semantic layer, the policy engine, agentic reasoning — inherits the latency and the failure modes decided here.

Why compute moves to the data: the MapReduce and Spark execution model

E1.1

The base asymmetry. A compiled job weighs kilobytes. A dataset weighs terabytes. So the platform never drags data to the code — it ships the code to the nodes where those bytes already sit on local disk. That is the locality principle MapReduce was built on and Spark inherited unchanged. The scheduler even ranks placement: PROCESS_LOCAL (same JVM) → NODE_LOCAL (same host) → RACK_LOCAL → ANY, and will wait a few hundred milliseconds for a better slot before accepting a worse one.

The unit of parallelism is the partition, not the row. A dataset is split into partitions; each partition gets exactly one task. Partition count is the ceiling on parallelism: 1 TB in 12 partitions is 12 tasks, and a 480-core cluster will sit 97% idle no matter how much memory you add.

Lazy evaluation and stage boundaries. Transformations execute nothing — they build a DAG. An action (count, write, collect) invokes the scheduler, which cuts that DAG into stages along dependency type:

• Narrow dependency: each output partition reads exactly one input partition — filter, select, withColumn. Nothing crosses the network, and several operators fuse into a single pass over the row.
• Wide dependency: an output partition reads many input partitions — groupBy, join, distinct, repartition. That is a shuffle: disk write, network transfer, and a mandatory stage boundary.

Why this dictates architecture. The next stage cannot start until the previous one finishes entirely, down to its last task. So there are exactly three levers: read less (columnar format, projection, predicates pushed into storage), shuffle less (aggregate before joining, broadcast the small side), and spread evenly (lesson E1.2).

What changed after classical MapReduce. MapReduce materialised every phase to the distributed filesystem, so iterative algorithms paid a full I/O round per iteration. Spark keeps intermediates in executor memory and, instead of replicating a lost partition, recomputes it from lineage. That is not "faster" — it is a different trade: cheap reuse, expensive failure late in a long DAG. Enterprise semantic layers rest on exactly this property; what Palantir Foundry describes as an ontology over the data fabric is a set of derived assets that must be able to rebuild themselves from source.

# Stage boundaries are cut at WIDE dependencies. Read the plan, not the docs.
orders = spark.read.parquet('s3a://lake/orders')        # 1.2 TB, 9,600 files

# NARROW: map-side only, no bytes cross the network, stays inside Stage 0
recent = (orders
          .filter(orders.order_ts >= '2026-01-01')
          .select('order_id', 'customer_id', 'total'))

# WIDE: groupBy forces an Exchange -> Stage 0 must fully finish first
per_customer = recent.groupBy('customer_id').sum('total')

per_customer.explain(mode='formatted')
# == Physical Plan ==
# * HashAggregate (final)                              <- Stage 1
# +- Exchange hashpartitioning(customer_id, 200)       <- SHUFFLE = stage cut
#    +- * HashAggregate (partial)                      <- Stage 0, runs on the data
#       +- * Project [order_id, customer_id, total]    <- column projection pushed
#          +- * FileScan parquet
#                PushedFilters: [GreaterThanOrEqual(order_ts, 2026-01-01)]

# Partition count = task count = ceiling on parallelism
print(recent.rdd.getNumPartitions())   # 3,100 -> 3,100 tasks on 480 cores

Partitioning and data skew: why 199 tasks wait for one

E1.2

Skew is not a bug — it is a property of real keys. A shuffle places rows by hash(key) % N. The hash is uniform over keys, not over rows: if 34% of the table carries the same customer_id, every one of those rows lands in one partition by definition. Raising the partition count changes nothing — the hot key hashes to the same place, it just gains more empty neighbours.

The symptom and how to read it. In the Spark UI look at the stage, not the job: task median 30 s, max 50 min. Then make the one comparison that separates two different diagnoses:

• The slow task read the same input bytes and rows as the median → not skew, a sick node (degraded disk, noisy neighbour, GC). Speculative execution fixes it.
• The slow task read 40× more rows → key skew. Speculation is useless: a copy of that same partition on another node still has to grind through the same 61 million rows.

Three working remedies, in order of cost.

• Adaptive Query Execution. The planner inspects real statistics at the stage boundary and splits any partition exceeding the median by skewedPartitionFactor into sub-partitions. Free in code terms, but sort-merge-join only, and it cannot save you from a single mega-key.
• Isolating the degenerate key. The most common enterprise skew is not a popular customer but a surrogate: NULL, -1, UNKNOWN, guest. Those rows join to nothing meaningful by definition — split them into their own branch and union the results.
• Salting. Append a random suffix 0..S-1 to the hot key and replicate the small side across all S salt values with explode. The key is artificially split into S partitions. The price: the small side grows S-fold, which is why S is tuned rather than set to 1000.

On-disk partitioning is a different decision. Do not conflate execution partitions with storage directories. Directory partitioning on event_date lets the planner prune whole folders before reading a byte. Partitioning on a high-cardinality key such as user_id produces millions of tiny files and destroys listing performance.

from pyspark.sql import functions as F

# 1) DIAGNOSE before you fix. What share of the table is the heaviest key?
(orders.groupBy('customer_id').count()
       .orderBy(F.desc('count')).show(5, False))
# +-----------+----------+
# |customer_id|count     |
# +-----------+----------+
# |-1         |784215330 |   <- 34% of 2.3B rows: the 'guest checkout' surrogate
# |4417019    |  1204880 |
# +-----------+----------+

# 2) Let the engine handle the moderate tail
spark.conf.set('spark.sql.adaptive.enabled', 'true')
spark.conf.set('spark.sql.adaptive.skewJoin.enabled', 'true')
spark.conf.set('spark.sql.adaptive.skewJoin.skewedPartitionFactor', '5')

# 3) Isolate the degenerate key - it joins to nothing anyway
real   = orders.filter(F.col('customer_id') != -1).join(dim, 'customer_id')
guests = orders.filter(F.col('customer_id') == -1).withColumn('segment', F.lit('GUEST'))
result = real.unionByName(guests, allowMissingColumns=True)

# 4) Salt what is still hot AFTER isolation. Both sides must agree on the salt.
S = 64
big   = orders.withColumn('salt', (F.rand() * S).cast('int'))
small = dim.withColumn('salt', F.explode(F.array([F.lit(i) for i in range(S)])))
joined = big.join(small, ['customer_id', 'salt'])   # small side is now 64x - budget for it

The cost of shuffle and the economics of caching

E1.3

Shuffle is the one operation that pays both prices. It writes the entire intermediate set to each executor's local disk (shuffle write) and then pulls it across the network to other executors (shuffle read). Everything else — scan, filter, project — pays only one. So the optimisation budget in this layer goes to shuffle, not to "faster code".

Post-shuffle partition size is the primary numeric lever. Target 128–200 MB per partition. The default spark.sql.shuffle.partitions = 200 was written for tens-of-gigabytes datasets; on 8 TB it yields ~40 GB per task, which guarantees spill and usually OOM. Too small is the opposite failure: 100,000 tasks of 2 MB each, where scheduling a task costs more than the work in it.

Spill is not a crash — it is an invoice. When a task's data exceeds its memory share it is written to disk and read back. The Spill (Memory) / Spill (Disk) metrics in the Spark UI are a direct measurement of how badly the partition size was chosen.

A broadcast join removes the shuffle entirely. If one side is smaller than autoBroadcastJoinThreshold (10 MB by default, in practice raised to 32–256 MB), it is shipped once to every executor and the big side joins locally — zero shuffle on the side holding the terabytes. It is the cheapest win in the layer. Its price is driver and executor memory, so the threshold is raised deliberately, not to the maximum.

Caching wins in exactly one case: the dataset is read more than once and recomputation costs more than the memory it will occupy. The rules that separate benefit from harm:

• Cache competes for the same memory as execution. 340 GB of cache against 288 GB of available memory is not a cache — it is guaranteed spill plus eviction.
• persist() is lazy: without an action to materialise it, the first use recomputes everything anyway.
• unpersist() is mandatory once the set is no longer needed, or it holds memory for the rest of the job.
• Cache does not truncate lineage. Against a long DAG with an expensive failure the right tool is checkpoint(), which materialises to reliable storage and cuts the dependency chain.

Load balancing is about evenness, not headcount. Adding executors helps only when there are enough partitions to occupy them and none is an order of magnitude heavier than the rest. Otherwise the new nodes simply wait alongside the old ones.

from pyspark import StorageLevel
from pyspark.sql import functions as F

# 1) Size the shuffle for the DATA, not for the default.
#    8 TB / 200 partitions = 40 GB per task -> spill, then OOM.
#    8 TB / 49,152 partitions ~= 170 MB per task -> inside the target band.
spark.conf.set('spark.sql.shuffle.partitions', 49152)

# 2) Broadcast the small side: the 18 TB fact table is never shuffled.
spark.conf.set('spark.sql.autoBroadcastJoinThreshold', 64 * 1024 * 1024)   # 64 MB
enriched = fact.join(F.broadcast(dim_sku), 'sku_id')                       # dim_sku = 41 MB

# 3) Cache ONLY what is genuinely read more than once, and materialise it once.
sessions = (enriched
            .filter(F.col('event_type').isin('view', 'add_to_cart', 'purchase'))
            .repartition('user_id')
            .persist(StorageLevel.MEMORY_AND_DISK))
sessions.count()                     # single materialisation pass

funnel  = sessions.groupBy('user_id').agg(F.collect_list('event_type').alias('path'))
revenue = sessions.filter(F.col('event_type') == 'purchase').groupBy('sku_id').sum('total')

funnel.write.mode('overwrite').parquet('s3a://lake/marts/funnel')
revenue.write.mode('overwrite').parquet('s3a://lake/marts/revenue')
sessions.unpersist()                 # give the memory back before the next stage

# 4) Long DAG with an expensive late failure? checkpoint(), not cache() -
#    only checkpoint truncates lineage.
spark.sparkContext.setCheckpointDir('s3a://lake/_checkpoints')
stable = enriched.checkpoint(eager=True)

Sources this area derives from

1. Spark - The Definitive Guide - Big data processing made simple
2. Martin-Kleppmann---Designing-Data-Intensive-Applications -O’Reilly-Media-(2017)
3. A Review on Big Data Analytics Framework
4. Handling Data Skew in MapReduce Cluster by Using P
5. A study of skew in mapreduce application
6. A Balanced Solution for the Partition ba
7. Characterizing and Fixing Silent Data Loss in Spark on AWS
8. DeltaLakeOptimizationTechniquesforScalableLakehouseArchitectures

Continue in this track