So far the course has focused on simple algorithms applied to Big Data — word count, secondary index, and the four design patterns of Chapter 4. Those all share a convenient property: each input record can be processed in isolation.
But what happens when the computation is complex?
Rethinking an algorithm from a sequential to a parallel logic is a challenge. It is not a matter of translating code. It usually means finding a different algorithm that computes the same answer through a data flow the framework can actually run, and then paying attention to how many passes over the data that flow costs.
This chapter works through two case studies. Prefix sum is the small one: an algorithm whose definition seems to forbid parallelism outright, and which yields to one good idea. Frequent itemset mining is the large one: a genuinely hard data mining problem with a decade of published parallel formulations, each fixing a specific weakness of the last. Read the second case study as a story about engineering trade-offs, because that is exactly what it is.
Two questions come before any design work. Does it always make sense to parallelize a sequential algorithm to take advantage of a Big Data architecture? And is it worth scaling out indefinitely?
Short answer: no.
Amdahl’s law: the performance gain depends on the actual percentage of the computation that can be parallelized. It is limited by the computation portion that cannot be parallelized. With P the parallel fraction and S the speedup due to parallelization:
1
real speedup = ─────────────────
(1 - P) + P / SAnd on top of that, you must account for the overhead in parallelization management — the scheduling, the shuffling, the task launch costs that Chapters 5 to 8 spent their time measuring.
Set the parallel fraction and the raw parallel speedup, and see the real speedup — and the ceiling you can never pass.
The examinable point is the asymptote. As S grows without bound, the term P/S vanishes and the real speedup tends to 1/(1−P). So the serial fraction alone fixes the maximum. Quote that alongside the reminder that parallelization management has its own overhead, which means the real curve does not merely flatten — past a point it turns back down.
Before the hard problems, the deck re-draws the easy one to establish a notation. Every algorithm from here on is presented as a chain of key-value shapes, with the operation that transforms one into the next written on the arrow.
Word count on sopra la panca la capra campa, sotto la panca la capra crepa:
| Stage | Content |
|---|---|
| INPUT | (0, sopra) (6, la) (9, panca) (15, la) (18, capra) (24, campa,) … |
| MAP | (sopra, 1) (la, 1) (panca, 1) (la, 1) (capra, 1) (campa, 1) … |
| SHUFFLE & SORT | (campa, [1]) (capra, [1,1]) (crepa, [1]) (la, [1,1,1,1]) … |
| REDUCE | (campa, 1) (capra, 2) (crepa, 1) (la, 4) … |
And the same thing in the design notation the rest of the chapter uses:
(offset, line)
│ flatMap
(offset, word)
│ map
(word, 1)
│ reduceByKey
(word, sum)Each box is the type of the RDD at that point, each arrow is a Spark operation from Chapter 6. Reading an algorithm design therefore means reading two things at once: what information is carried at each step, and — from the arrow labels — where the shuffles are. Every reduceByKey, groupByKey or join in these diagrams is a stage boundary and a network cost. That is the whole basis on which the competing Apriori formulations later in the chapter are judged.
Given a sequence of numbers x0, x1, x2, … return a sequence y0, y1, y2, … where
y_i = Σ x_j
j ≤ i| X | 1 | 2 | 3 | 4 | 5 | 6 |
|---|---|---|---|---|---|---|
| Y | 1 | 3 | 6 | 10 | 15 | 21 |
Now look at what the definition demands: each item requires the previous items to compute the result. That is the exact opposite of the property that made word count easy. Every output depends on every input before it, so there is no obvious way to hand a slice of the problem to a machine that cannot see the other slices.
How can this be parallelized? (Assume the input to be ordered.)
The trick is to stop thinking about items and start thinking about partitions. The five steps from the deck:
Run the five steps on the example from the slides and watch the local results turn global.
Assuming ordered input:
(partitionID, orderedNumbers[])
│ map
(partitionID, sum)
│ Bjoin( green.partitionID <= blue.partitionID )
(blue.partitionID, green.sums[])
│ map
(partitionID, PS)Assuming unordered input (and assuming max is known), two operations are prepended to rebuild the ordering:
(offset, number)
│ map
(partitionId, number)
│ groupByKey
(partitionId, orderedNumbers[])
│ map
(partitionId, sum)
│ Bjoin( green.partitionId <= blue.partitionId )
(blue.partitionId, green.sums[])
│ map
(offset, PS)Note the Bjoin — a broadcast join. The set of partition totals is tiny (one number per partition), so it can be sent to every executor rather than shuffled. This is the same reasoning as the Broadcast Hash join of Chapter 7 and the broadcast variables of Chapter 8: when one side of a join is small, replicating it everywhere is cheaper than moving the big side. The pattern compute local, broadcast the summary, correct locally recurs in almost every algorithm in this chapter.
Now the serious case study. The vocabulary first:
| Term | Definition |
|---|---|
| Itemset | A set of items, e.g. {Milk, Bread, Diaper} |
| k-itemset | A set of k items |
| Transactions | Occurrences of one or more items |
| Support of an itemset | The fraction of transactions that contain the itemset |
The running example, five transactions:
| TID | Items |
|---|---|
| 1 | Bread, Milk |
| 2 | Bread, Diaper, Beer, Eggs |
| 3 | Milk, Diaper, Beer, Coke |
| 4 | Bread, Milk, Diaper, Beer |
| 5 | Bread, Milk, Diaper, Coke |
So s({Milk, Bread, Diaper}) = 2/5 — only transactions 4 and 5 contain all three.
A frequent itemset is an itemset whose support is higher than a certain threshold, with 0 < minsup ≤ 1. The problem is to find all frequent itemsets, independently of their length.
The naive approach is to match every candidate against each transaction, and it is way too complex:
Given d items, the total number of itemsets is 2d. Not 2d operations for some clever algorithm — 2d candidates to consider. With a supermarket catalogue of a thousand products, the lattice has more nodes than there are atoms within reach. No amount of cluster is going to fix an exponential; the algorithm has to avoid generating most of the space.
The escape from the exponential is one observation, published by Agrawal and Srikant (Fast Algorithms for Mining Association Rules in Large Databases, VLDB 1994, pp. 478-499):
If a k-itemset is frequent, then all (k−1)-itemsets are also frequent.
This follows the anti-monotone property of support: ∀X, Y : (X ⊆ Y) ⇒ s(X) ≥ s(Y).
Put another way — and this is the form you actually use: if a k-itemset is infrequent, then all (k+1)-itemsets are also infrequent.
The contrapositive is what does the work. The moment an itemset is found infrequent, its entire supersets subtree in the lattice can be deleted without ever being counted. Pruning one node near the top of the lattice removes an exponential number of nodes below it.
With threshold = 3/5 (i.e. a count of at least 3 out of 5 transactions), the 1-itemset counts are:
| 1-itemset | Count | Verdict |
|---|---|---|
| Bread | 4 | frequent |
| Coke | 2 | infrequent |
| Milk | 4 | frequent |
| Beer | 3 | frequent |
| Diaper | 4 | frequent |
| Eggs | 1 | infrequent |
Coke and Eggs are out — and with them, every itemset that contains them. The slide says it in a parenthesis worth memorising: “No need to generate those involving Coke or Eggs”. Four frequent items remain, so there are C(4,2) = 6 candidate 2-itemsets instead of the C(6,2) = 15 the full lattice would hold. Of those six, four clear the threshold, and they in turn admit exactly 1 candidate 3-itemset, {Bread, Milk, Diaper}, whose support is 2/5 — below the threshold, so the algorithm stops.
The arithmetic that summarises the whole idea:
| Strategy | Itemsets considered |
|---|---|
| If every subset is considered | 6C1 + 6C2 + 6C3 = 6 + 15 + 20 = 41 |
| With support-based pruning | 6 + 6 + 1 = 13 |
The counts for the 2- and 3-itemsets printed on this slide come out garbled in the extracted text (the columns interleave), so the per-itemset counts above have been recomputed directly from the five transactions on the same slide. They agree with the support the deck states explicitly earlier, s({Milk, Bread, Diaper}) = 2/5, and with the 6 + 6 + 1 = 13 arithmetic the slide concludes with. If your copy of the slide shows different counts in that middle table, trust the transaction table.
Note the property that will dominate every parallel version: it requires several passes on the transaction set.
k = 1
repeat
generate candidate itemsets based on frequent k-1 itemsets
for each transaction
for each candidate itemset
increase itemset's support
end for
end for
keep frequent k-itemsets
until k = |d| or no more frequent k-itemsetsRun the algorithm on the five transactions with threshold 3/5. Watch candidates get generated, counted, and pruned — and watch the candidates that are never generated at all.
Two observations open the door:
So the shape of the solution is already familiar. What the next five sections argue about is not whether to use map and reduce, but how many passes over the transaction set to pay, and how many key-value pairs to generate on the map side.
The most direct translation (Li & Zhang, The strategy of mining association rule based on cloud computing, ICBCGI 2011, pp. 475-478):
| Summary | Discussion |
|---|---|
|
|
(TID, items[])
│ flatMap
(itemset, 1)
│ reduceByKey
(itemset, count)
│ filter
(itemset, count)Map-side, the algorithm computes the partial support for every possible itemset — the entire lattice, lit up for each transaction. Reduce-side, it computes the final support for every possible itemset, and only then filters.
This design throws away the Apriori principle entirely. A transaction with n items emits 2n itemsets, and the combiner can only compress what has already been generated — it cannot un-generate it. One pass over the data sounds like the best possible outcome until you notice that the pass produces an exponential amount of intermediate data, which then has to be shuffled. This is Chapter 8’s lesson in a new costume: the pass count is not the only cost that matters.
PApriori (Li, Ning, et al., Parallel implementation of apriori algorithm based on mapreduce, SNPD/ACIS 2012) restores the principle by restoring the levels:
| Summary | Discussion |
|---|---|
|
|
run(k = 1)
↓
(TID, items[])
│ flatMap
(k-itemset, 1)
│ reduceByKey
(itemset, count)
│ filter
(itemset, count) ──→ save
│
continue? ──→ run(k++)The first job computes partial support for 1-itemsets on the map side, then final support and filtering on the reduce side. The second job does the same for candidate 2-itemsets, the third for candidate 3-itemsets, and so on — with each level restricted to the candidates the previous level certified.
Be able to state the trade-off between the first two designs in one sentence each. One-pass FIM reads the transaction set once but generates the whole lattice per transaction. PApriori generates only surviving candidates, but re-reads the entire transaction set once per level — and in Hadoop MapReduce every one of those levels means a fresh job, writing to and reading from HDFS (Chapter 5). The next two designs each attack one of those two costs.
PApriori’s repeated passes are exactly the pathology Spark was built to remove.
(Qiu, Hongjian, et al., Yafim: a parallel frequent itemset mining algorithm with spark, IEEE IPDPS Workshops 2014.)
Performance gain: 1 order of magnitude with respect to PApriori.
Two lines of Chapter 6 and 8 machinery buy a factor of ten. cache() means the k passes read RAM instead of HDFS; the broadcast of the candidate set means every executor gets the current level’s candidates without a shuffle. The algorithm did not change — the logic is still PApriori. Only the data movement did.
(Rathee, Sanjay, Manohar Kaul, and Arti Kashyap, R-Apriori: an efficient apriori based algorithm on spark, PIKM/CIKM 2015.)
It effectively drops execution time of the second iteration — less space requirements, and less time to generate candidates on the map side.
Why the second iteration specifically? Because that is where the candidate set is largest: after 1-itemset pruning there are still many surviving items, and the number of pairs is quadratic in that count. Level 2 is the bulge in the lattice, so a data structure that is cheaper to consult and smaller to broadcast pays off there and nowhere else.
(Yahya, Othman, Osman Hegazy, and Ehab Ezat, An efficient implementation of A-Priori algorithm based on Hadoop-Mapreduce model, International Journal of Reviews in Computing 12, 2012; Farzanyar, Zahra, and Nick Cercone, Efficient mining of frequent itemsets in social network data based on MapReduce framework, ASONAM 2013.)
These abandon the level-by-level structure for something cleverer:
The whole design rests on one implication, stated twice in the slides:
If the itemset’s support is never locally frequent (in any partition), it cannot be globally frequent. Equivalently: if the itemset is globally frequent, it has to be locally frequent somewhere.
The reasoning is a pigeonhole argument. If an itemset fell below the threshold fraction in every partition, then summing over partitions it must fall below the threshold overall. So the union of the locally frequent sets is a superset of the globally frequent sets — a candidate list with no false negatives, produced in a single distributed pass. The second job then counts exactly that list, which is why no accuracy is lost.
(partitionID, items[]) (partitionID, items[])
│ flatMap(apriori on partition) │ flatMap
(itemset, partialCount) (itemset, 1)
│ filter & distinct │
(localFrequentItemset) ──────── join ─────────────→ │
(itemset, 1)
│ reduceByKey
(itemset, count)
│ filter
(itemset, count)Phase one, map side: local Apriori to detect locally frequent itemsets; reduce side: simply output them. Phase two, map side: compute partial support for all candidate itemsets; reduce side: compute their final support.
Discussion: a different approach to parallelization, which performs better than k-passes.
The mapper receiving the whole split rather than one transaction is a real change of unit, and it is the same move as the prefix sum in section 5: process the entire partitions, not the single items. It also imports a risk. Local Apriori must run inside one task, so the split has to fit the memory and time budget of a single task — and any skew in how transactions are distributed (Chapter 8) now translates directly into a straggler running a full sequential mining algorithm.
The last design (Zhang, Feng, et al., A distributed frequent itemset mining algorithm using Spark for Big Data analytics, Cluster Computing 18.4, 2015, pp. 1493-1501) changes the representation rather than the schedule.
The worked example from the slide:
i 0 0 1 1 0 1 1 0 1 s({i}) = 5
j 1 0 0 1 1 0 0 1 1 s({j}) = 5
s({i,j}) = 2AND the two vectors position by position and you get 0 0 0 1 0 0 0 0 1: two ones, hence support 2. The counting has become a bitwise operation.
Toggle the bits of the two item vectors and watch the supports recompute. This is the DFIMA example from the slides, made editable.
Discussion: IO cost reduced due to caching, but boolean vectors must fit memory.
The constraint is the examinable half. A boolean vector has one bit per transaction, so its length grows with the size of the dataset, and there is one vector per frequent item. The technique converts a data-scan problem into a memory-residency problem — which is a good trade when the transaction count is moderate and a fatal one when it is not. Every design in this chapter is a trade of exactly this kind, which is the point of studying five of them.
A single map-reduce job. The mapper computes every possible itemset in the transaction, the combiner computes partial supports, the reducer computes the final supports.
Naive implementation; too many key-value pairs generated on the map-side, even with combiner.
Chain of map-reduce jobs. Each job computes the support of k-itemsets for a given k, and the frequent k-itemsets are passed as input to the next mapper.
Better than one-pass; many passes on the transaction set; still many key-value pairs generated on the map-side.
Spark version of PApriori (k-passes). The transaction set is cached; at each iteration the frequent k-itemsets are collected and broadcasted in a HashTree.
Performance gain of 1 order of magnitude with respect to PApriori.
Improved version of YAFIM. Replaces the HashTree with a BloomFilter, only for the map-side generation of 2-itemsets.
Effectively drops execution time of the second iteration: less space requirements, less time to generate candidates on the map-side.
Two phase map-reduce jobs. The first mapper applies sequential Apriori to the local split (its input is the whole split, not the single transaction); the first reducer outputs the list of candidate frequent itemsets; the second job restarts from scratch and calculates the support of the locally frequent itemsets.
A different approach to parallelization; performs better than k-passes.
Similar to a two phase map-reduce job, but the transaction set is cached. First compute the frequent 1-itemsets and their boolean vectors, which are broadcasted; the support of a k-itemset is determined by the logical AND between the items’ vectors. Then use the vectors to compute frequent k-itemsets.
IO cost reduced due to caching, but boolean vectors must fit memory.
No. Amdahl’s law says the performance gain depends on the actual percentage of the computation that can be parallelized, and is limited by the portion that cannot be. With P the parallel fraction and S the speedup due to parallelization, the real speedup is 1 / ((1−P) + P/S). As S grows the real speedup tends to 1/(1−P), so the serial fraction alone fixes a ceiling that no cluster size can pass. On top of that you must account for the overhead in parallelization management, which means past a certain point adding nodes makes things worse rather than merely failing to help.
Because each item requires the previous items to compute the result: yi is the sum of all xj with j ≤ i, so every output depends on every earlier input. The solution changes the unit of work: divide input into partitions; process the entire partitions, not the single items; compute partial prefix sums within each partition; save and broadcast the maximum value from each partition; compute final prefix sums by adding to each partial the required offsets. The sequential dependency is not removed, it is lifted to the level of partitions, where only a handful of numbers need to be combined.
It joins each partition with the sums of all partitions whose id is less than or equal to its own — Bjoin(green.partitionID <= blue.partitionID) — so that each partition can compute the offset to add to its local prefix sums. A broadcast join is right because the joined side is tiny: one number per partition. Replicating it to every executor costs far less than shuffling the data itself, which is the same reasoning behind the Broadcast Hash join in Chapter 7 and broadcast variables in Chapter 8.
An itemset is a set of items; a k-itemset is a set of k items; transactions are occurrences of one or more items. The support of an itemset is the fraction of transactions that contain it — for example s({Milk, Bread, Diaper}) = 2/5 in the five-transaction example. A frequent itemset is one whose support is higher than a threshold minsup, with 0 < minsup ≤ 1, and the problem is to find all frequent itemsets independently of their length. The naive approach matches every candidate against each transaction and is way too complex: given d items, the total number of itemsets is 2d.
If a k-itemset is frequent, then all (k−1)-itemsets are also frequent. Equivalently and more usefully: if a k-itemset is infrequent, then all (k+1)-itemsets are also infrequent. It follows from the anti-monotone property of support: for all X, Y, if X ⊆ Y then s(X) ≥ s(Y). It is due to Agrawal and Srikant (VLDB 1994). The second form is what makes the algorithm efficient: pruning one infrequent node deletes its entire superset subtree from the lattice without those nodes ever being generated or counted.
If every subset is considered the count is 6C1 + 6C2 + 6C3 = 6 + 15 + 20 = 41 itemsets. With support-based pruning it is 6 + 6 + 1 = 13. The saving comes from the two infrequent 1-itemsets — Coke (count 2) and Eggs (count 1) — because as the slide notes there is then no need to generate those involving Coke or Eggs. Four frequent items leave C(4,2) = 6 candidate pairs instead of 15, and those admit exactly one candidate 3-itemset, {Bread, Milk, Diaper}, whose support of 2/5 falls below the threshold.
First, itemsets are inferred from the items in a transaction, and each transaction is independent from the others — so a map function can return the itemsets in the given transaction. Second, calculating the support of an itemset means counting its occurrences, which is pretty much like word counting, so the reduce side is a familiar sum-by-key. What the competing designs then argue about is not the shape but the cost: how many passes over the transaction set, and how many key-value pairs on the map side.
A single map-reduce job: the mapper computes every possible itemset in the transaction, the combiner computes partial supports, the reducer computes the final supports, and a filter keeps the frequent ones. The design is naive and generates too many key-value pairs on the map-side, even with a combiner. The deeper problem is that it discards the Apriori principle: a transaction of n items emits 2n itemsets, and a combiner can only compress data that has already been generated. One pass over the input is not a win if that pass produces an exponential amount of intermediate data to shuffle.
PApriori is a chain of map-reduce jobs where each job computes the support of k-itemsets for a given k, and the frequent k-itemsets are passed as input to the next mapper — so candidate generation is restricted level by level, restoring the Apriori pruning. It is better than one-pass, but it makes many passes on the transaction set and still generates many key-value pairs on the map-side. In Hadoop MapReduce each level is a separate job, so every pass means writing to and reading from HDFS.
YAFIM is the Spark version of PApriori, still k-passes. Two changes: the transaction set is cached, so the repeated passes read memory instead of HDFS; and at each iteration the frequent k-itemsets are collected and broadcasted in a HashTree rather than shuffled. The gain is one order of magnitude over PApriori. Note that the algorithmic logic is unchanged — only the data movement is.
It is an improved YAFIM that replaces the HashTree with a BloomFilter, only for the map-side generation of 2-itemsets. It effectively drops execution time of the second iteration thanks to less space requirements and less time to generate candidates on the map side. The second iteration is the right target because it is where the candidate set is largest: after 1-itemset pruning many items survive, and the number of candidate pairs is quadratic in that number, so level 2 is the bulge in the lattice.
Two phase map-reduce jobs. The first mapper applies sequential Apriori to the local split — its input is the whole split, not the single transaction — and the first reducer simply outputs the list of candidate frequent itemsets. The second map-reduce job restarts from scratch and calculates the support of the locally frequent itemsets. Correctness rests on the principle stated in the slides: if an itemset’s support is never locally frequent in any partition, it cannot be globally frequent, and conversely if an itemset is globally frequent it has to be locally frequent somewhere. So the union of locally frequent sets is a superset of the globally frequent ones — a candidate list with no false negatives. It is a different approach to parallelization and performs better than k-passes.
It is similar to a two phase map-reduce job but the transaction set is cached. First it computes the frequent 1-itemsets and, for each, a boolean vector where Vi[t] = 0 if item i is missing in transaction t and 1 otherwise; these vectors are broadcasted. The support of a k-itemset is then determined by the logical AND between the items’ vectors — in the slide example, two vectors each with support 5 give s({i,j}) = 2. Then vectors are used to compute frequent k-itemsets. IO cost is reduced due to caching, but the boolean vectors must fit memory: each vector has one bit per transaction and there is one per frequent item, so the technique converts a data-scan problem into a memory-residency problem.
That no design wins on every axis. Each moves cost from one resource to another: One-pass FIM trades intermediate data for passes; PApriori trades passes for pruning; YAFIM spends memory (caching and broadcasting) to remove HDFS round-trips; R-Apriori spends a smarter data structure to shrink the widest level; MRA/IMRA spends local computation and a per-split memory budget to reduce the pass count to two; DFIMA spends memory on boolean vectors to turn counting into a bitwise operation. Choosing among them is a question about your data — how many transactions, how many items, how much RAM per executor — not a question with a single right answer.