Skip to content
academia.sh
tr
Log in

Course Advanced

Distributed Data Processing

Published Updated

By the end of this course

Start course

01

Distributed Computing

Building the basic abstractions for splitting a job across several machines on a fictional cluster and counting the cost of each one: the driver's plan, the executor's share and the definition of a task, with a single-worker baseline giving a makespan of 2250 and a total work of 2200; the case where block placement and replication make a read local, and the number of blocks stored as the replication factor grows; which step of the map and reduce abstraction stays local and which one touches the network; splitting rows into partitions and lazy evaluation eliminating computation that is never used; the number of tasks a resource manager makes wait while balancing the number of workers against concurrent tasks, and the fixed cost of fifty units for handing work to each worker.

  1. 01 Cluster Computing Fundamentals Splitting a job across several machines with the driver, executor and task abstractions, and measuring the split with two separate numbers: the 2200 rows coming out of the pipeline give a makespan of 2250 and a total work of 2200 on a single worker, but when split across four workers by the subscriber key the makespan drops to 611, when split by the region key two of the four workers stay idle, the skew is 3.32 instead of 1.02 and the makespan can only fall to 1876; because the total work of the three setups is exactly the same 2200, a speedup claim that does not state which one was measured is incomplete.
  2. 02 Distributed File Systems Splitting data into blocks, placing them on the cluster, and measuring whether the worker that processes a block already holds it: when 2200 rows are split into 22 blocks of 100 rows, with a single copy and assignment by block order only 5 of the 22 blocks are read locally, the makespan is 1850 and the total work is 5600; at a replication factor of 2 the local blocks rise to 17 and the makespan falls to 1150, at 3 they rise to 22 and the makespan falls to 650 but the stored blocks grow from 22 to 66, and a fourth copy adds nothing to reading; locality-aware assignment reads 22 of 22 blocks locally with a single copy, but because the placement is unbalanced the makespan stays at 750 instead of 650.
  3. 03 MapReduce Model Splitting a computation into a step that works independently per row and a step that sums per key, and measuring that only one of the two steps touches the network: the map processes 2200 rows locally and moves 0 rows, the makespan is 611 and the total work stays at 2200, while the reduce step that produces the same result moves 1738 rows from the subscriber layout to the region key and raises the total work to 7414 and the makespan to 1914; when every worker reduces its own share before moving it, the moved rows are only 15 of 20 intermediate rows, the total work falls to 2245 and the makespan to 622, and the five region totals presented by the three setups remain exactly the same.
  4. 04 Distributed Datasets A distributed dataset consisting of partitions and a plan that has not run yet, and counting the computation that lazy evaluation eliminates: while eleven partitions are spread over four workers as three, three, three and two, the eager setup opens 44 partitions, reads 6556 rows and writes 3727 intermediate rows for a result of five rows, the lazy setup gives the same result by opening a single partition and reading 200 rows, leaving 2000 rows never read; in a result of a hundred rows the rows read triple while the makespan stays at 250 because three partitions run on three separate workers; in the count action that asks for the whole result the lazy setup reads 2200 rows and the difference comes only from the unused branch and from the intermediate rows not written; because building the plan reads no rows, the type error in the plan appears only at the first action and after 200 rows have been read.
  5. 05 Resource Managers Counting that the number of tasks and the number of concurrent tasks are two separate knobs in scheduling a job on a cluster, and where the fixed cost of fifty units per task goes: with one task in one slot the makespan is 2250 and the fixed cost 50, while sixteen tasks in sixteen slots bring the makespan down to 215 but raise the fixed cost to 800; increasing concurrency does not change the fixed cost at all, whereas increasing the number of tasks raises it by fifty units per task; splitting the job into more tasks at the same concurrency raises the makespan from 611 to 950; going from eight tasks to sixteen the makespan gain stays at 121 and loses to the 400 units of added fixed cost; and when two jobs run on the same cluster the priority and the shared policy change the order of finishing but do not change the 600 units of fixed cost paid.

02

Performance

Counting where a distributed job slows down and when attempts to speed it up make it worse: shuffling in the same layout moving 1738 rows and raising the makespan from 611 to 1914 and the total work from 2200 to 7414, the broadcast join winning up to a size of 1600 rows while the fact side is partitioned by region and the crossover point lying between 1600 and 4000, the skew of 1.02 on the subscriber key rising to 3.32 on the region key with two of four workers left idle so that adding workers becomes meaningless, the total of two queries falling from 5500 to 3300 when the intermediate result is cached, lowering the checkpoint interval from six to three leaving the rows recomputed at the same failure point at 400, the makespan of 2250 on a single worker rising to 3691 with a second worker, and the total work rising from 5830 to 8635 while the makespan falls, which makes it mandatory to write which number was measured.

  1. 01 Shuffle Cost Counting for the first time where a distributed job slows down: when the same 2200 rows laid out by the subscriber key are repartitioned by the subscriber key, the shuffled rows stay at 0, the makespan is 611 and the total work 2200; when the same rows are shuffled to the region key, 1738 rows cross the network, the makespan rises to 1914 and the total work to 7414, and all three setups produce exactly the same result. When the network cost is swept over 1, 3, 6 and 12, the shuffled rows do not change at all and the 611 and 2200 of the setup without a shuffle stay fixed, while the makespan of the setup with a shuffle moves from 1045 to 5825 and its total work from 3938 to 23056; the ordering of the three keys comes out the same at all four values, so the ordering is insensitive to the choice and the magnitude is entirely of the choice. The cost of a shuffle comes not from the computation itself but from moving the data from its place.
  2. 02 Joins at Scale Measuring a job that brings two sets face to face with two separate strategies and counting the point where the winner changes: while the fact side is partitioned by the region key, the broadcast join copies the dimension side to every worker and writes a makespan of 1926 at a dimension of 50 rows, 2276 at 400 and 3476 at 1600; the partitioned join, however, has to move 1738 rows of the fact side no matter how small the dimension gets, so it cannot drop below 3179. Because the makespan of the broadcast join grows by one unit per dimension row and that of the partitioned join by nine sixteenths of a unit, the two lines cross: the makespans are equal at 2978 and 2979, the partitioned join pulls ahead at 2980, and at 4000 the gap is 5876 against 5429. The same crossover point comes out as 2980 when total work is measured. The decision is sensitive to the network cost: at a cost of 1 the crossover point drops to 535, and at 6 and 12 the partitioned join cannot pull ahead at any size. Both strategies match the same rows at every size.
  3. 03 Data Skew Measuring an unbalanced partitioning and counting why adding workers is not a fix: when the same 2200 rows are partitioned by the subscriber key, the load of the four workers stays between 539 and 561 and the skew comes out as 1.02; when they are partitioned by the region key, the load becomes 374, 0, 0 and 1826, the skew rises to 3.32 and two workers receive no rows at all. When the number of workers is increased with the wrong key, the skew rises from 1.66 to 3.32, 6.64 and 6.88 and the number of idle workers from 0 to 13; the makespan, which is 2250 on a single worker, rises to 3691 with two workers and drops below the baseline only at sixteen workers, while the total work is carried from 2200 to 8635. With the right key, no rows are shuffled at any number of workers, the total work stays fixed at 2200 and the makespan falls to 215. All three partitionings produce exactly the same region totals.
  4. 04 Caching and Persistence Counting the cost of computing the same intermediate result twice when more than one query reads it, and the cost that comes back when the cache is not held: two queries sharing an intermediate result of 2200 rows total 5500 without a cache and 3300 with one, but with a single query both are 2750 and the cache gains nothing. The intermediate result sits in the memory of four workers as 539, 550, 561 and 550 rows; as workers fail, the total rises from 3300 to 3850, 4411, 4961 and 5500, and a cache that is not held at all pays exactly the cost of the setup without a cache, although the plan was made for 3300. The failure of a single worker moves the total to between 3839 and 3861. The same cache shows a gain of 40% in the mixed total, 25% in total work and 23.91% in makespan, and a claim of speedup is incomplete unless it is written which of the three numbers was measured.
  5. 05 State in Stream Engines Counting how much a failed worker asks to have back in a job that flows continuously, and whether raising the checkpoint frequency really lowers that number: in a streaming job of 2200 rows, when a failure occurs in period 2026-09, 1600 rows are recomputed without a checkpoint, 400 at interval 6, again 400 at interval 3, and 0 at interval 1. Halving the interval does not halve the recovery, because the last point before the failure has not moved; interval 4 gives 0 on the same failure. When all eleven failure points are scanned, interval 6 makes 545.5 rows on average be recomputed and interval 3 makes 218.2, but the certain cost rises from 400 to 800; by the worst case interval 4 and 3 tie, by the average interval 4 wins. When the failure probability drops from 1 to 0.5, the best choice turns back to holding no checkpoint. With partitioned state a single failed worker asks for 408 rows instead of 1600 and 102 instead of 400, and exactly-once requires holding exactly that many rows of state.
  6. 06 Cost and Cluster Sizing Counting that the decision of how many workers a cluster is set up with is made between makespan and total work, and that the two numbers go in opposite directions: the same job of 2200 rows gives a makespan of 2250, 3691, 3179, 2610 and 1398 and a total work of 2200, 5830, 7414, 8074 and 8635 on one, two, four, eight and sixteen workers; as the makespan falls the total work rises, and the sizing decision cannot be made without writing which one is paid. The second worker raising the makespan from 2250 to 3691 comes from the shuffle coming into being, and even with a good key of skew 1.00 a rise from 2250 to 2800 is seen. The product of workers and makespan climbs one way from 2250 to 22368, when the fixed cost is counted per worker the gain of sixteen workers falls from 852 to 102, and when the network cost rises to 12 the setup with the shortest makespan returns to a single worker. The lesson closes the course and gathers the numbers of eleven lessons in one table.

Final exam

Distributed Data Processing — Final Exam

In preparation

This course's question bank is not complete yet. The exam opens here automatically once enough questions are added.

Start typing to search.

↑↓ ↵ Esc navigate · open · close