MRCall+SMS.py

def map_entries(index, entry):
    return (entry.number, entry)


def reduce_entries(key, entries):
    for i in range(0, len(entries)) :
        entry = entries[i]
        if entry.type == "call":
            sms = find_sms(entries,
                           entry.number_to,
                           i+1,
                           entry.time,
                           entry.time + 5*60)
            if sms != None:
                yield (entry,sms)
        

def find_sms(entries, number_to, start, time_start, time_end):
    for i in range(start, len(entries)) :
        entry = source[i]
        if (entry.type == "sms" and
            entry.number_to == number_to and
            entry.time > time_start and
            entry.time < time_end) :
            return entry
        elif entry.time > time_end:
            return None
# Run Map
for i in range(0, len(source)) :
    key, entry = map_entries(i, source[i])
    sorted_map[key].append(entry)

# Sort the map(keys and values)
sorted_map.sort()

# Run Reduce
for k in sorted_map.keys():
    for (call, sms) in reduce_entries(k, sorted_map[k]):
        found_items.append(entry+sms)