Sort-Merge Join on user_id — Joining Two Sharded Datasets in MapReduce (Figures 11-2 / 11-3)
left: the fact and dimension tables · middle: the shuffle gathers one user's records · right: the reducer and the next job
activity events (clickstream)
user_id url timestamp
42 /news 10-01 09:12
7 /blog 10-01 11:40
42 /docs 10-01 10:05
42 /news 10-02 08:20
· · ·
fact table of the star schema — sharded into multiple
files, so many mappers process it in parallel
user profiles (dimension)
user_id date_of_birth
42 1990-07-14
7 1985-02-03
· · ·
dimension table of the star schema — also sharded,
read row by row by its own mapper
mapper over events:
emit (key=user_id, value=url)
mapper over profiles:
emit (key=user_id, value=DOB)
shuffle by user_id
sorted reducer input for user_id = 42
(42, dob=1990-07-14)
profile
(42, url=/news, 10-01 09:12)
(42, url=/docs, 10-01 10:05)
(42, url=/news, 10-02 08:20)
secondary sort: profile record first,
then the user's events in timestamp order
reducer — sort-merge join
dob = first value
# the profile record
for e in events:
# sorted by timestamp
emit(e.url, dob)
out: (url, date_of_birth) per page view
Why it is a sort-merge join
one user's record in memory at a time;
no network calls during the join;
mapper output is sorted by key; reducers
merge the sorted lists from both sides
next MapReduce job — group by + aggregation
shuffle by url: the reducer sees all page views for one URL
it keeps a counter of views per age group, one URL at a time
emit (url, age_group, view_count)
group by + aggregation — viewer age comes from date_of_birth
pill = pipeline stage · box = record or code · dashed = why the join works · arrow = data flow