def map(line):
    values = line.split(",")
    return (values[0], float(values[-1]))

def reduce(year, temperatures_in_year):
    return (year, sum(temperatures_in_year) / len(temperatures_in_year))

Since we can have up to one measurement per second, the maximum number of measurements \(M_{max}\) for a certain year is given by the following formula:

\[ M_{max} = 365 \times 24 \times 60 \times 60 \approx 31.5 \times 10^6 \]

Since there might be up to 31 million values associated with a key, the bottleneck of the computation would be the shuffle operation, since we need to copy a high number of (key,value) pairs from the mappers to the reducers.

Also, a reducer might have to loop over a huge list of values in order to compute their average.

Function map() is the same as in the previous exercise.

def map(line):
    values = line.split(",")
    return (values[0], float(values[-1])) 

Now we code the function combine() which will apply some computation on key-value pairs returned by the Map task. More precisely, after a Map task applies the function map() on all lines of a block, a local shuffle operation is executed to group all values associated with the same key. The function combine() takes in a couple, of which the key (a year) is associated with the list of all its values (the year’s temperatures). The function returns a couple, where the key still represents a year and the value is itself a couple; the first element of this couple is the sum all of year’s temperatures, the second element is the number of temperatures in the year. Please note that all these considerations are applied to data within a single bloc. Therefore, the shuffle and the combine did not move data across blocks. The result is that from each block we send out to the reducers only a couple per year, which significantly reduces the amount of data sent over the network.

def combine(year, temperatures_in_year):
    return (year, (sum(temperatures_in_year), len(temperatures_in_year)))

The function reduce() takes in a a couple, where the key is a year and the value is a list of couples, as output by the function combine().

def reduce(year, sum_len_tuples):
    sum_temps = sum(s for (s, _) in sum_len_tuples)
    nb_temps = sum(l for (_, l) in sum_len_tuples)
        
    return (year, sum_temps/nb_temps)

The result of our algorithm is a collection of couples, where the first element represents a couple of individuals, and the second element is the common friends of the two individuals.

We infer that a couple of individuals is a proper choice for a key.

Now, the function map() only takes in a single line. The question we need to ask ourselves is: can we obtain a solution to our problem on that single line? Otherwise stated, can we still get common friends between couple of individuals from one single line?

Take the example of line: A,D,B,C. From this line, we know that A is a common friend of his friends : (B, D), (C, D) and (B, C).

This gives us an idea as to the implementation of the function map().

def map(line):
    values = line.split(",")
    return [((x, y), values[0]) for x in values[1:] for y in values[1:] if x < y] 

Note that the condition x<y guarantees that we never output summetrical couples.

The shuffle step will then associate the list of common friends to each couple of individuals. This is exactly what we want, therefore there is no need for a reduce() function (identity function).

The second equation is more appropriate because it allows the computation of the sum of the elements and of the square of the elements step by step by using map and combine together.

Instead, if we use the first equation, we need first to compute the average and then use it to compute the variance.

def map(line):
    values = line.split(",")
    return (values[0], float(values[-1])) 

def combine(year, temperatures_in_year):
    return (year, (sum(temperatures_in_year), sum(x**2 for x in temperatures_in_year), len(temperatures_in_year)))

def reduce(year, sum_len_tuples):
    sum_temps = sum(s for (s, _, _) in sum_len_tuples)
    sum_sq_temps = sum(sq for (_, sq, _) in sum_len_tuples)
    nb_temps = sum(l for (_, _, l) in sum_len_tuples)
        
    return (year, (sum_temps/nb_temps, math.sqrt(sum_sq_temps/nb_temps - (sum_temps/nb_temps)**2)))