1. alex,Cupertino,5
2. jane,Cupertino,8
3. jeff,Cupertino,6
4. jeff,Cupertino,8
5. alex,Sunnyvale,10
6. jane,Sunnyvale,8
7. jeff,Sunnyvale,3
8. alex,Stanford,8
9. jane,Stanford,8
10. jeff,Stanford,10

------------------------
11. jeff,Stanford,10
12. bob,Sunnyvale,10
13. rafa,Sunnyvale,10
14. jane,Sunnyvale,4
15. jane,Sunnyvale,10
16. david,Ames,8
17. alex,Ames,10
18. jane,Ames,10
19. jeff,Ames,10
20. jeff,Saratoga,8
21. alex,Saratoga,1
22. jane,Saratoga,7
---------------------------
23. jeff,Saratoga,4
24. alex,Stanford,8
25. jane,Stanford,8
26. jeff,Stanford,10
27. jeff,Stanford,10
28. bob,Sunnyvale,10
29. rafa,Oakland,34
30. jane,Oakland,4
31. jane,Oakland,1
32. jane,Fremont,6
33. jane,Fremont,3
34. jane,Fremont,1

Mapper:
map(key, value) {
    # tokenize the given input record
    info = value.split(",")
    person_name = info[0] 
    city = info[1]   
    rating = int(info[2])
    
    # filter based on rating
    if rating in range(2, 11) {
        emit(city, (rating, 1))
    }
}

output:
partition 1: 
alex,Cupertino,5 => (Cupertino, (5, 1))
jane,Cupertino,8 => (Cupertino, (8, 1))
jeff,Cupertino,6 => (Cupertino, (6, 1))
jeff,Cupertino,8 => (Cupertino, (8, 1))
alex,Sunnyvale,10 => (Sunnyvale, (10, 1))
jane,Sunnyvale,8 => (Sunnyvale, (8, 1))
jeff,Sunnyvale,3 => (Sunnyvale, (3, 1))
alex,Stanford,8 => (Stanford, (8, 1))
jane,Stanford,8 => (Stanford, (8, 1))
jeff,Stanford,10 => (Stanford, (10, 1))

partition 2:
jeff,Stanford,10 => (Stanford, (10, 1))
bob,Sunnyvale,10 => (Sunnyvale, (10, 1))
rafa,Sunnyvale,10 => (Sunnyvale, (10, 1))
jane,Sunnyvale,4 => (Sunnyvale, (4, 1))
jane,Sunnyvale,10 => (Sunnyvale, (10, 1))
david,Ames,8 => (Ames, (8, 1))
alex,Ames,10 => (Ames, (10, 1))
jane,Ames,10 => (Ames, (10, 1))
jeff,Ames,10 => (Ames, (10, 1))
jeff,Saratoga,8 => (Saratoga, (8, 1)) 
alex,Saratoga,1 => Ignore (rating < 2)
jane,Saratoga,7 => (Saratoga, (7, 1))

partition 3:
jeff,Saratoga,4 => (Saratoga, (4, 1))
alex,Stanford,8 => (Stanford, (8, 1))
jane,Stanford,8 => (Stanford, (8, 1)) 
jeff,Stanford,10 => (Stanford, (10, 1)) 
jeff,Stanford,10 => (Stanford, (10, 1)) 
bob,Sunnyvale,10 => (Sunnyvale, (10, 1)) 
rafa,Oakland,34 => Ignore (rating > 10) 
jane,Oakland,4 => (Oakland, (4, 1)) 
jane,Oakland,1 => Ignore (rating < 2) 
jane,Fremont,6 => (Fremont, (6, 1)) 
jane,Fremont,3 => (Fremont, (3, 1)) 
jane,Fremont,1 => Ignore (rating < 2)

Sort and Shuffle:
# Sort phase
(Cupertino, (5, 1))
(Cupertino, (8, 1))
(Cupertino, (6, 1))
(Cupertino, (8, 1))

(Sunnyvale, (10, 1))
(Sunnyvale, (8, 1))
(Sunnyvale, (3, 1))
(Sunnyvale, (10, 1))
(Sunnyvale, (10, 1))
(Sunnyvale, (4, 1))
(Sunnyvale, (10, 1))
(Sunnyvale, (10, 1))

(Stanford, (8, 1))
(Stanford, (8, 1))
(Stanford, (10, 1))
(Stanford, (10, 1))
(Stanford, (8, 1))
(Stanford, (8, 1))
(Stanford, (10, 1))
(Stanford, (10, 1))

(Ames, (8, 1))
(Ames, (10, 1))
(Ames, (10, 1))
(Ames, (10, 1))

(Saratoga, (8, 1))
(Saratoga, (7, 1))
(Saratoga, (4, 1))

(Oakland, (4, 1))

(Fremont, (6, 1))
(Fremont, (3, 1))

# Final Output
(Cupertino, [(5, 1), (8, 1), (6, 1), (8, 1)]) 
(Sunnyvale, [(10, 1), (8, 1), (3, 1), (10, 1), (10, 1), (4, 1), (10, 1), (10, 1)]) 
(Stanford, [(8, 1), (8, 1), (10, 1), (10, 1), (8, 1), (8, 1), (10, 1), (10, 1)]) 
(Ames, [(8, 1), (10, 1), (10, 1), (10, 1)]) 
(Saratoga, [(8, 1), (7, 1), (4, 1)]) 
(Oakland, [(4, 1)]) 
(Fremont, [(6, 1), (3, 1)])

reducer:
# key: city_name<String>
# values: Iterable<Tuple>, Tuple -> (Integer, Integer)

reduce(key, values) {
    unique_cities_10 = set()  # to keep track of unique cities with rating 10
    sum = 0
    count = 0
    for r, c in values {
        # (r, c) = (rating, count) 
        if r == 10 {
            unique_cities_10.add(key)
        sum += r
        count += c
        }
    }
    # Aggregation
    avg_rating = sum / count
    
    if unique_cities_10 {
        emit("cities_with_rating_10", list(unique_cities_10))
    }

    if avg_rating >= 4 {
        emit(key, avg_rating)
    }
}

output:
("cities_with_rating _10", [Sunnyvale, Stanford, Ames])
(Cupertino, 6.75)
(Sunnyvale, 8.13)
(Stanford, 9.00)
(Ames, 9.50)
(Saratoga, 6.33)
(Oakland, 4.00)
(Fremont, 4.50)