1、Big Data Analytics with Parallel Jobs,Ganesh Ananthanarayanan University of California at Berkeley http:/www.cs.berkeley.edu/ganesha/,Rising philosophy of data-ism,Diagnostics and decisions backed by extensive data analytics“In God we trust. Everybody else bring data to the table.“ Competitive and S
2、ocial BenefitsDichotomy: Ever-increasing data size and ever-decreasing latency target,Parallelization,Massive parallelization on large clustersData growing faster than Moores lawComputation Frameworks,Dryad,Computation Frameworks,Job DAG of tasks Outputs of upstream tasks are passed downstreamJobs i
3、nput (file) is divided among tasks Every task reads a block of dataComputation and storage are co-located Tasks scheduled for locality,Task2,Task1,Task3,Computation Frameworks,Block,Slot,Promise of Big Data Analytics,Goal: Efficient execution of jobs Completion time and utilizationChallenge: Scale D
4、ata size and parallelism Performance variations stragglers,Efficient and fault-tolerant execution of parallel jobs on large clusters,IO intensive Memory,Tasks are IO intensiveTask completes faster if its input is read off memory instead of disk Memory local taskFalling memory prices 64GB/machine at
5、FB in Aug 2011, 256GB/machine not uncommon now,Can we move all data to memory?,Proposals for making “RAM as the new disk” (e.g., RAMCloud)Huge discrepancy between storage and memory capacities 200x more data on disks than available memory,Use memory as cache,Production traces Dryad and Hadoop cluste
6、rs Thousands of machines Over 1 million jobs in all, span of 6 months (2011-12),Will the inputs fit in cache?,Will the inputs fit in cache?,Heavy-tailed distribution of jobs sizes 92% of jobs can fit their inputs in memory,We built a memory cache,Simple in-memory distributed cache Cache input data o
7、f jobsSchedule tasks for memory localitySimple cache replacement policies Least Recently Used (LRU) and Least Frequently Used (LFU),We built a memory cache,Replayed the Facebook trace of Hadoop jobsJobs sped up by 10% with LFU; hit-ratio of 67% Beladys MIN (optimal) 13% improvement with hit-ratio of
8、 74%,How do we make caching really speedup parallel jobs?,Parallel jobs require a new class of cache replacement algorithms,Parallel Jobs,Tasks of small jobs run simultaneously in a wave,Task duration (uncached input),Task duration (cached input),All-or-Nothing: Unless all inputs are cached, there i
9、s no benefit,All-or-Nothing for multi-waved jobs,Large jobs run tasks in multiple waves Number of tasks is larger than number of slots Wave-width: Number of parallel tasks of a job,time,slot5,slot4,slot3,slot2,slot1,completion time,All-or-Nothing for multi-waved jobs,Large jobs run tasks in multiple
10、 waves Number of tasks is larger than number of slots Wave-width: Number of parallel tasks of a job,time,slot5,slot4,slot3,slot2,slot1,completion time,All-or-Nothing for multi-waved jobs,time,slot5,slot4,slot3,slot2,slot1,Large jobs run tasks in multiple waves Number of tasks is larger than number o
11、f slots Wave-width: Number of parallel tasks of a job,All-or-Nothing for multi-waved jobs,time,slot5,slot4,slot3,slot2,slot1,Large jobs run tasks in multiple waves Number of tasks is larger than number of slots Wave-width: Number of parallel tasks of a job,Cache at the wave-width granularity,Cache a
12、t the wave-width granularity,Job with 50 tasks, wave-width of 10,All-or-nothing,How to evict from cache?,View at the granularity of a jobs input (file) Evict from incompletely cached waves Sticky Policy,With Sticky Policy,slot1slot2slot3slot4slot5slot6 a slot7slot8,completion,Hit-ratio: 50% No speed
13、-up of jobs,Job 1,Job 2,Without Sticky Policy,Which file should be evicted?,Depends on metric to optimize:User centric metric Completion time of jobsOperator centric metric Utilization of the cluster,What are the eviction policies for these metrics?,Reduction in Completion Time,Idealized model for j
14、ob: Wave-width for job: W Frequency predicts future access: F Task duration is proportional to data read: D Speedup factor for cached tasks: ,Cost of caching: W D Benefit of caching: Benefit/cost: F/W,Completion Time of Job: frequency/wave-width,F,D,How to estimate W for a job?,Job size,Wave-width (
15、slots),Use the size of a file as a proxy for wave-width Relative ordering remains unchanged,Improvement in Utilization,Idealized model for job: Wave-width for job: W Frequency predicts future access: F Task duration is proportional to data read: D Speedup factor for cached tasks: ,Cost of caching: W
16、 D Benefit of caching: D F Benefit/cost: F,Utilization of job: frequency,W,Isnt this just Least Frequently Used?,All-or-Nothing property matters for utilization Inter-dependent tasks overlap Reduce tasks start before all map tasks finish (to overlap communication),All-or-Nothing No wastage!,Cache Ev
17、iction Policies,Completion time: LIFE (Largest Incomplete File to Evict) Evict from file with lowest (frequency/wave-width) Sticky: fully evict a file before next (all-or-nothing)Utilization policy: LFU-F (Least Frequently Used File granularity) Evict from file with the lowest frequency Sticky: full
18、y evict a file before next (all-or-nothing),How do we achieve the sticky policy?,Caches are distributed Blocks of files are stored across machines How do we know which files are incomplete?Coordination Global view of all the caches which blocks to evict (sticky policy) where to schedule tasks (memor
19、y locality),PA Man: Centralized Coordination,Global view,DFS: Distributed File System,Evaluation Setup,Workload derived from Facebook & Bing tracesPrototype in conjunction with HDFSExperiments on 100-node EC2 cluster Cache of 20GB per machine,Sticky Policy,Reduction in Completion Time,Improvement in
20、 Utilization,Sticky Policy,PA Man: Scalability,PACMan Coordinator Near-uniform latency of 1.2ms with up to 11,000 clients/s,PACMan Client Can scale up to 20 simultaneous tasks on the machine,Coordination infrastructure is sufficiently scalable,How far is LIFE from the optimal for speeding up paralle
21、l jobs?,Caches for parallel jobs are complex!,Maximize # of “job hits” (a la Beladys MIN) Job hit = 1 if all inputs are cached; = 0 otherwise Cache space: Aggregated vs. Distributed,vs.,Caches for parallel jobs are complex!,Maximize # of “job hits” (a la Beladys MIN) Job hit = 1 if all inputs are ca
22、ched; = 0 otherwise Cache space: Aggregated vs. Distributed Partial overlap of input blocks between jobs,Job 1,Job 2,Caches for parallel jobs are complex!,Maximize # of “job hits” (a la Beladys MIN) Job hit = 1 if all inputs are cached; = 0 otherwise Cache space: Aggregated vs. Distributed Partial o
23、verlap of input blocks between jobs Inputs: Equal-sized vs. varying wave-width,Blocks,Jobs,Simplifying Assumptions,Cache space: Aggregated vs. Distributed At 100% locality, aggregated distributed Assume: Aggregated cache spacePartial overlap of input blocks between jobs 98% of jobs have no partial o
24、verlap Assume: No partial overlapInputs: Equal-sized vs. varying wave-width,Notation,Jobs indexed by 1, 2, , n Job i has wave-width wi and frequency fi Total cache space = CSingle-waved jobs,Problem Definition,Job i has wave-width wi and frequency fi Total cache space = C xi : if input of job i is i
25、n cache Maximize # “job hits”Maximize i fi xiSubject to i wi xi Cxi = 0, 1 Solution to 0-1 knapsack is optimal cache configuration,Linear Relaxation,*Discarded input is one with largest wave-width,0 xi 1,xi = 0, 1,Truncation Discard fractional item from linear relaxation,0-1 Knapsack,Theorem: Favori
26、ng (Frequency/wave-width) is (1-1/K) optimal, where K = C/(maxi wi).,* (frequency/wave-width) is optimal solution,Comparing “job hits” (i fi xi),Theorem: LIFE + Optional Eviction is (1-1/K) optimal, where K = C/(maxi wi).,Optional eviction = can evict input of incoming job(unlike traditional caches)
27、,LIFE + Optional Eviction,LIFE + Optional Eviction,Most jobs are small (heavy-tailed distribution) K and (1-1/K) 1; LIFE + Optional Eviction Optimal,Optional Eviction: Practical improvement to LIFE,All-or-nothing property for parallel jobs Cache at granularity of wave-widthsPACMan: Coordinated in-me
28、mory caching systemCache replacement policies for completion time (LIFE) and utilization (LFU-F); not hit-ratioOptimality analysis; close approximation,PA Man Highlights,Straggler Mitigation,Stragglers: slow tasks that delay job completion All-or-Nothing job as slow as the slowest task Mantri (for l
29、arge production jobs) First work to present insights from the wild Cause-aware solution; in deployment at Bing Dolly (for small interactive jobs) Clone jobs upfront; pick first to finish Heavy-tailed job sizes mitigate nearly all stragglers with just 3% extra resources,Promise of Big Data Analytics,
30、Efficient and fault-tolerant execution of parallel jobs“PACMan: Coordinated Memory Caching for Parallel Jobs”, NSDI12 “Scarlett: Coping with Skewed Popularity Content in MapReduce Clusters”, EuroSys11 “True Elasticity in Multi-Tenant Data-Intensive Compute Clusters”, SoCC12 “Disk-Locality in Datacen
31、ter Computing Considered Irrelevant”, HotOS11“Reining in Outliers in MapReduce Clusters using Mantri”, OSDI10 “Effective Straggler Mitigation: Attack of the Clones”, NSDI13 “Why let resources idle? Aggressive Cloning of Jobs with Dolly”, HotCloud12,Caching & Locality,Straggler Mitigation,onclusion,All-or-Nothing property of parallel jobsPA Man: Coordinated Cache Management Provably optimal cache replacement Straggler mitigation Cause-based solution for large jobs, cloning for small jobs,
