Programming for Data Science at Scale
Distributed Key-Value
Processing
Amir Shaikhha, Fall 2025
Key-Value Pairs
• Single-node
– Key-value pairs = Dictionaries
• Dictionaries are not the most commonly
collections in single-node programs
• List/Arrays are most common
2
Distributed Key-Value Pairs
• Most common in big data processing
• Key design choice in MapReduce
– Manipulating key-value pairs
3
4
Distributed Key-Value Pairs
• Large datasets are often made up of
complex nested data records
• To work with such datasets, it is often
desirable to project down these nested
data types into key-value pairs
5
JSON Example
• It may be disrable to create:
// city: String
val properties: RDD[(String, Property)]
case class Property(
street: String,
city: String,
state: String)
• Where instances of this
RDD are grouped by their
cities
6
Pair RDDs
• Often when working with distributed data,
it’s useful to organize data into key-value
pairs.
• In Spark, distributed key-value pairs are
called Pair RDDs.
• Pair RDDs allow you to
– Act on each key in parallel
– Regroup data across the network
7
Pair RDDs
• Such RDDs are treated specially by Spark
• Spark automatically adds a number of
useful additional methods
def groupByKey(): RDD[(K, Iterable[V])] = ...
def reduceByKey(f: (V, V) => V): RDD[(K, V)] = ...
def join[W](other: RDD[(K, W)]): RDD[(K, (V,W))] = ...
8
Creating Pair RDDs
• Pair RDDs are most often created from
existing non-pair RDDs
val rdd: RDD[Property] = ...
val pairRDD: RDD[(String, String)] =
[Link](p => ([Link], [Link]))
• Once created, you can use Pair-RDD-
specific transformations
9
Transformations on Pair RDD
• groupByKey
• reduceByKey
• mapValues
• keys
• join
• left0uterJoin/right0uterJoin
10
Grouping in Scala Collections
• Recall groupBy of Scala collections
class List[T] {
def groupBy[K](f: T => K): Map[K, Traversable[T]]
}
• Partitions this collection into a map of
traversable collections according to some
descriminator function
11
Scala Collections Example
• Let's group the below list of ages into
"child", "adult", and "senior" categories.
val ages = List(2, 52, 44, 23, 17, 14, 12, 82, 51, 64)
val grouped = [Link]({age =>
if (age >= 18 && age < 65) "adult"
else if (age < 18) "child"
else "senior"
})
//grouped: [Link][String,List[Int]] =
//Map(senior-> List(82), adult-> List(52, 44, 23, 51, 64),
//child-> List(2, 17, 14, 12))
12
Grouping in Pair RDDs
• Spark Pair-RDDs’ groupByKey
– A groupBy on Pair RDDs specialized on
grouping all values that have the same key
– Thus, no argument is required
class PairRDD[K, V] {
def groupByKey(): RDD[(K, Iterable[V])]
}
13
Spark Grouping Example
case class Event(organizer: String,
name: String,
budget: Int)
val eventsRdd = [Link](...)
.map(event => ([Link], [Link]))
val groupedRdd = [Link]()
• If the key is organizer, what does this call
do?
• Nothing ☺
14
Spark Grouping Example
case class Event(organizer: String,
name: String,
budget: Int)
val eventsRdd = [Link](...)
.map(event => ([Link], [Link]))
val groupedRdd = [Link]()
[Link]().foreach(println)
// (Prime Sound,CompactBuffer(42000))
// (Sportorg, CompactBuffer(23000, 12000, 1400))
// ...
15
Reduction in Pair RDDs
• Conceptually, reduceByKey can be
thought of as a combination of
groupByKey and reduce-ing on all the
values per key.
• It's more efficient though, than using each
separately.
class PairRDD[K, V] {
def reduceByKey(f: (V, V) => V): RDD[(K, V)]
}
16
Pair RDD Reduction Example
case class Event(organizer: String,
name: String,
budget: Int)
val eventsRdd = [Link](...)
.map(event => ([Link], [Link]))
val budgetsRdd = [Link](_ + _)
[Link]().foreach(println)
// (Prime Sound, 42000)
// (Sportorg, 36400)
// ...
17
Other Pair RDD operations
• mapValues[U](f: V => U): RDD[(K, U)]
– Can be thought of a short-hand for:
[Link] { case (x, y)=> (x, f(y))}
– Simply applies a function to only the values in
a Pair RDD
• countByKey(): Map[K, Long]
– Action
– Count the number of elements per key in a
Pair RDD
18
Pair RDD Averaging Example
// Calculate a K-V pair containing (budget, #events)
val intermediate = [Link] (b => (b, 1) )
.reduceByKey((vl, v2) => (vl._1 + v2._1, vl._2 + v2._2))
val avgBudgets = [Link] {
case (budget, numberOfEvents) => budget / numberOfEvents
}
[Link]().foreach(println)
// (Prime Sound, 42000)
// (Sportorg, 12133)
// ...
19
Joins on Pair RDDs
• Another transformation on Pair RDDs.
• They're used to combine multiple datasets.
RDD
Join RDD
RDD
class PairRDD[K, V] {
def join[W](other: RDD[(K, W)]): RDD[(K, (V,W))]
}
20
Joins on Pair RDDs
• They are one of the most commonly-used
operations on Pair RDDs!
• What happens to the keys when two
RDDs don’t contain the same key
• Two kinds
– Inner joins (join)
– Outer joins (leftOuterJoin / rightOuterJoin)
21
Pair RDD Join Example
case class Event(organizer:String,name:String,budget:Int)
case class Organizer(id: String, revenue: Int)
val eventsRdd = [Link](...)
.map(event => ([Link], [Link]))
val orgRdd = [Link](...)
.map(org => ([Link], [Link]))
val joinRdd = [Link](orgRdd)
[Link]().foreach(println)
// (Prime Sound, (42000, 2000000))
// (Sportorg, (23000, 4000000))
// (Sportorg, (12000, 4000000))
// (Sportorg, (1400, 4000000))
// ...
22
Other operations
23
[Link]
References
• Compulsory reading:
• MapReduce [OSDI’04]:
• MapReduce: Simplified data processing on large clusters
• Spark [NSDI’12]
• Resilient distributed datasets: A fault-tolerant abstraction for in-
memory cluster computing
• Recommended reading:
• YARN [SoCC’13]
• The next generation of M/R
• Dryad [EuroSys’07]
• Generalized framework for data-parallel computations
24
QUESTIONS?
25