ShoreKeeper: Accelerating Iterative Data Analytics in the Cloud via Distributed Recovery-Cost-Aware Caching

Challenges

• Node-local caching can evict blocks while remote memory remains free, or discard blocks that are expensive to recover across the cluster.Uneven block sizes and computation times can saturate some nodes earlier than others, even when nodes run the same operations.Source: §1 · p. 1
• Preemptions can destroy costly cached blocks, while existing cloud-native responses migrate random local blocks to idle remote memory.ShoreKeeper responds to preemption alarms by reoptimizing migration, and periodically replicates a minimal set of costly shuffle-produced blocks to handle missed alarms.Source: §1 · p. 2

ShoreKeeper

• ShoreKeeper uses a global MILP to choose which partitions to cache and where to place them, minimizing execution time under time-varying cluster memory capacity.It maps distributed dataflows into one computational graph that captures cluster-wide recovery costs and cross-node transfer latency.Source: §1 · p. 2
• Graph pruning restricts optimization to blocks relevant to a near-future window, bringing solver decision latency below 4 seconds in experiments.For blocks assigned to another node, transfers run in the background using network capacity idle during foreground computation.Source: §1 · p. 2

Evaluation

• ShoreKeeper achieves up to 6.11× speedup over cost-agnostic caching and up to 2.05× over Blaze on on-demand instances, and up to 2.87× over Blaze with Spark’s Decommissioner on preemptible instances.The largest gains occur on workloads with the highest cross-executor imbalance; cost-aware evacuation and proactive replication protect high-cost partitions on preemptible instances.Source: §1 · p. 2

Dataflow execution

• A distributed dataflow DAG represents intermediate RDDs as vertices and transformations as edges; iterations are jobs divided into stages and executed by parallel tasks on executors.Later jobs consume predecessor outputs, so intermediate partitions may be reused across iterations.Source: §2 · p. 2

Recovery costs

• Discarding a partition can require recursively recomputing missing ancestors along its lineage, making one cache miss much more costly than a single transformation.Recovery cost depends on whether each partition along the lineage remains in memory; spilled partitions instead return through disk I/O.Source: §2 · pp. 2–3

Cost-aware caching

• Blaze unifies cost factors in an ILP objective that minimizes total potential recovery cost for partitions accessed by upcoming iterations, subject to memory capacity, balancing caching benefits against recovery costs.This approach outperforms cost-agnostic heuristics by up to 2.86×.Source: §2 · p. 3

3.1 Recovery cost imbalance

• Runtime filtering, aggregation fan-in, and shuffle redistribution can undo initially balanced partitions, propagating uneven sizes into uneven computation times.These runtime factors are not anticipated by offline partitioning.Source: §3.1 · p. 3
• Recovery costs shift among executors across iterations in both PageRank and SVD++, while locality-aware scheduling, load balancing, and remote fetching trade off memory, compute, and network costs.Lower-cost executors can absorb partitions from heavier peers, but prior redistribution methods do not account for caching or recovery costs.Source: §3.1 · pp. 3–4

3.2 Deployment Environments

• Preemptible instance pools change compute and memory capacity during execution, and a departing node loses its accumulated cached partitions.In the cited 32-instance AWS Spot trace, about 25% of instances were preempted within 30 minutes; losing one instance in the final iteration increased completion time by 30–50% for evaluated workloads.Source: §3.2 · p. 4
• Spark’s node decommissioner migrates cached partitions at random and is constrained by peers’ idle memory; missed alarms bypass decommissioning and lose the departing node’s cached data.The trace’s 0.3% alarm miss rate motivates a lightweight proactive approach rather than periodic checkpointing for such rare events.Source: §3.2 · p. 5

4 ShoreKeeper Overview

• The system manages aggregate memory as a unified pool across on-demand VMs or preemptible instances, with minimal replication available for missed alarms.Its current design focuses on clusters provisioned with a single resource type.Source: §4 · p. 5

4.1 Design Overview

• ShoreKeeper jointly decides whether to cache, where to place, and whether to replicate partitions, using recovery cost to guide cluster-wide plans.Transfers are asynchronous and use network capacity left idle by foreground computation.Source: §4.1 · p. 5

4.2 System Architecture

• At each iteration boundary, the cost model and MILP solver produce a cluster-wide plan that the Runtime Manager turns into executor directives.Executor reports at stage boundaries keep partition sizes, computation times, and cache states current; arrivals and preemption alarms trigger replanning.Source: §4.2 · pp. 5–6

5.1 Profiling

• Offline dry-run and bandwidth profiling initialize partition, lineage, and cluster estimates; online profiling updates them as execution and cluster composition change.Node departures and arrivals trigger a bandwidth benchmark for the updated cluster.Source: §5.1 · p. 6

5.2 Cost Modeling

• The cost model represents partitions as cached, spilled to origin disk, or discarded, and prices access by the cheapest available location while recursively accounting for recomputation.When a partition is unavailable, its recovery cost sums the costs of its parents; the model fixes its target executor to the prior iteration's assignment.Source: §5.2 · pp. 6–7

5.3 Solver Design

• ShoreKeeper minimizes aggregate recovery cost across executors, subject to memory and placement constraints, while leaving scheduler decisions outside the solver.The fixed-assignment cost is a proxy for prioritizing partitions, not an exact prediction of recovery time under a particular scheduler.Source: §5.3 · p. 7, §5.2 · p. 7
• Cached-block boundary pruning limits decisions to lineage between next-iteration targets and nearest cached partitions, while shuffle approximation removes all-to-all edges and assigns cumulative shuffle cost to outputs.Together these techniques leave narrow dependencies and make lineage size scale as O(V).Source: §5.3 · p. 8

• The Runtime Manager diffs each new plan against the current one and sends only changed partition actions, while stage-completion reports keep cache and cost state current.Stage-level reporting balances responsiveness to mid-iteration changes against the round trips of per-task reports.Source: §6 · p. 8
• When an executor joins, ShoreKeeper profiles its network and disk throughput, replans, and can redistribute partitions to use its capacity immediately.This avoids waiting for task placement to populate the new executor's cache.Source: §6 · p. 8
• On preemption alarms, partitions are evacuated in descending recovery-cost order; unalarmed departures lose local cache, with optional replicas preserving shuffle partitions.Without replicas, lost partitions are recovered through lineage recomputation when needed.Source: §6 · p. 8

• The Spark implementation modifies approximately 8k lines across storage and scheduling, with the master coordinating cluster-wide decisions and aggregating partition metrics for the solver.Executor-side managers map solver outputs to cache, disk, transfer, or eviction actions.Source: §7 · p. 8
• The solver runs as a standalone Python process using Gurobi 11.0.3 and communicates with Spark's JVM master through JSON over named pipes.The master sends a compressed DAG, memory capacities, and bandwidth metadata, then receives block-to-executor assignments.Source: §7 · p. 9
• Lazy memory release retains remotely transferred partitions until local memory is exhausted, then frees removable partitions in ascending recovery-cost order.Transfers are asynchronous, batched by destination, and pipelined; they pause during shuffle stages that saturate network bandwidth.Source: §7 · p. 9

8.1 Experimental Setup

• Evaluation covered four graph and machine-learning benchmarks on 11 AWS r5n.2xlarge instances, with 20 executors sharing 318 GB of aggregate Spark memory.Baselines included LRU, LRC, MRD, Blaze, and Flint; preemption-aware comparisons also included Blaze+D and Blaze+DF.Source: §8.1 · p. 9

8.2 Deployments on VM Instances

• 2.05–6.11×, 1.20–3.03×, 1.01–2.40×, and 1.19–2.99× speedup over baselines in PR, CC, LR, and SVD++, respectively.PR had the highest cross-executor imbalance and deepest lineage, and ShoreKeeper achieved its largest gain there.Source: Figure 7 · p. 10, §8.2 · p. 10

8.3 Deployments on Preemptible Instances

• 1.14–2.87× speedup over Blaze+D across workloads and preemption windows; on PR, gains were 2.87× under Low and 2.5× under High.ShoreKeeper retained its advantage under High, when multiple instances were preempted simultaneously and the grace period allowed little time to react.Source: Figure 9 · p. 10, §8.3 · pp. 10–11

8.2 Deployments on VM Instances

• On PR, Blaze outperformed MRD by 1.8×, but ShoreKeeper overcame per-executor optimization limits by coordinating placement across executors.CC’s shorter lineage left less room for cluster placement; LR’s working set fit per-executor memory, while SVD++ partitions carried high recovery costs.Source: §8.2 · p. 10

8.4 Sensitivity Studies

• At 30 seconds ShoreKeeper retained its advantage over Blaze+D; shorter grace periods penalized it more, while slower networks still left it ahead.ShoreKeeper also remained roughly 2.1× faster than Blaze from 10 to 40 executors; execution time increased 10.7% as the system scaled.Source: Figure 10 · p. 11, §8.4 · p. 11
• At 6,768 partitions per RDD, pruning kept solver time below 4 seconds, compared with over 30 seconds without pruning at 423 partitions.Across four workloads, pruning kept the recovery-cost objective within 2% of the unpruned objective.Source: Figure 11 · p. 11, §8.4 · p. 11

8.5 Design Validation

• Under Low preemption, unified caching halved PR execution time and accounted for around 80% of total speedup; under High, all three components contributed substantially.Cost-aware decommission and proactive replication were necessary to cope with frequent preemptions under High.Source: Figure 12 · p. 11, §8.5 · p. 12
• Full S3 disaggregation took 2.02× longer and incurred 66.9% higher estimated cost than caching over ten PR iterations.The comparison used a 100M-vertex graph and excluded S3 request fees; CPU overhead was 1.4–2.4% for profiling, below 2.7% for the solver, and below 0.05% for migration threads.Source: §8.1 · p. 9, §8.5 · p. 12

Deployment Scenarios

• All-at-once reclamation of a purely preemptible cluster leaves ShoreKeeper to recompute from the original input; adding a few on-demand nodes is a remedy.For on-demand clusters, hardware failures and autoscaling departures are handled by the same Runtime Manager mechanisms.Source: §9 · p. 12

Profiling Overhead

• Profiling startup costs can matter for short jobs, although they amortize across long iterative jobs; sampling or reusing profiles could reduce the overhead.Offline profiling provides initial estimates that ShoreKeeper refines online during execution.Source: §9 · p. 12

Generality

• ShoreKeeper requires immutable materialized partitions, a dependency graph, and recomputation from ancestors; engines including Flink, Ray, and Dask could be backends with modest engineering effort.Extending beyond iterative workloads is harder because iteration boundaries limit the solver to the next iteration; without them, it must reason over the entire dataflow.Source: §9 · p. 12

Table 1

• Table 1 covers execution time and cost for ten PageRank iterations; its full-S3 total includes $0.6241 storage and excludes EBS and S3 request fees.The caption does not provide table cell values.Source: Table 1 · p. 12

Distributed Memory Management

• Remote-memory systems expose or harvest cluster memory, whereas ShoreKeeper places intermediate data across changing nodes to reduce recovery cost.The passage names FaRM, soNUMA, Infiniswap, Fastswap, AIFM, Semeru, and Carbink among these systems.Source: §10 · p. 12

Adapting to Resource Changes

• Systems such as Pado, ExoSphere, Accordion, BigQuery shuffle, POLARIS, and on-demand state separation adapt execution or state to resource changes, but treat state as equally worth preserving.ShoreKeeper instead retains partitions according to lineage-aware recovery costs.Source: §10 · p. 12

Checkpoint-Based Recovery

• Checkpoint-based systems trade storage and I/O cost for bounded recovery time, while ShoreKeeper keeps intermediate data in memory and optimizes caching, placement, and replication as capacity changes.The passage lists Flint, TR-Spark, SpotOn, Tachyon, and cost-based approaches.Source: §10 · p. 12

External Storage and Analytical Caching

• Analytical caches and disaggregated stores change data location or access; ShoreKeeper selects reusable partitions as executor capacity changes, so the approaches operate at different layers and can be combined.The passage gives Crystal, AnyBlob, Pocket, Locus, Jiffy, and Tachyon/Alluxio as examples.Source: §10 · p. 13

Conclusion

• Managing recovery costs across cloud executors is essential for reducing execution time in iterative workloads.ShoreKeeper addresses recovery-cost imbalance and resource preemption across diverse cloud deployment scenarios.Source: §11 · p. 13
• ShoreKeeper separates planning from execution, using MILP to balance caching benefits against eviction costs across all in-memory cluster partitions.Source: §11 · p. 13
• The design addresses recovery-cost imbalance across executors and resource preemption to support diverse cloud deployment scenarios.Source: §11 · p. 13