Word Count and Multiset Intersect Using Only a Mock Spark RDD API

You are given a tiny, single-machine imitation of a Spark RDD. Using only the methods it provides, implement two new functions on top of it.

from collections import defaultdict

class SimpleRDD:
 def __init__(self, items):
 self.items = items

 def flatMap(self, func):
 """Apply func to each item, flatten the results, and return a new SimpleRDD"""
 new_items = []
 for item in self.items:
 new_items.extend(func(item))
 return SimpleRDD(new_items)

 def groupByKey(self):
 """Group (key, value) pairs by key and return a SimpleRDD of (key, [values])"""
 grouped = defaultdict(list)
 for key, value in self.items:
 grouped[key].append(value)
 return SimpleRDD(list(grouped.items()))

 def union(self, other_rdd):
 """Return a new SimpleRDD containing items from self and other_rdd"""
 return SimpleRDD(self.items + other_rdd.items)

 def collect(self):
 """Return a list copy of the RDD items"""
 return list(self.items)

Note that the class has no map, filter, reduceByKey or distinct.

Clarifying Questions

• May a function call collect() and finish the work in plain Python, or must the result be built by chaining the provided transformations?
• Must the result keep a particular order, or is any order of the output items acceptable?
• Are the input items always hashable, given that groupByKey uses keys as dictionary keys?

Part 1 — Word count

Implement word_count(rdd), which takes an RDD of sentences and returns an RDD that pairs each word with the number of times it appears.

sentences = SimpleRDD([
 "hello world",
 "hello spark",
 "hello world"
])
wc_rdd = word_count(sentences)
print(wc_rdd.collect()) # Output: [('hello', 3), ('world', 2), ('spark', 1)]

Clarifying Questions for this Part

• Are sentences split into words on whitespace only, or also on punctuation?
• Are "Hello" and "hello" the same word?

What This Part Should Cover

• Expressing the missing map step with the provided methods
• Turning grouped values into a count per word
• Returning a SimpleRDD, and the order in which its items come out

Part 2 — Intersect

Implement intersect(rdd1, rdd2), which takes two RDDs and returns an RDD of the items they have in common.

rdd1 = SimpleRDD([1, 3, 3])
rdd2 = SimpleRDD([3, 4, 3])
result = intersect(rdd1, rdd2)
print(result.collect()) # Output: [3, 3]

The expected output keeps the repeated 3, so this is not a plain set intersection.

Clarifying Questions for this Part

• If a value appears twice in rdd1 and three times in rdd2, how many copies belong in the result? The example (two copies on each side, two in the result) does not settle it.
• Should the result follow the order of rdd1?

What This Part Should Cover

• Duplicate semantics that match the example, stated before coding
• Combining both inputs while keeping track of which side each item came from
• Time and memory cost, including a value that appears a very large number of times