<jupyter_start><jupyter_text>Map Reduce Algorithm Map reduce is very functional algorithm where three parts of it can easily executed on different machines. In this assignment, we will try to implement this algorithm into a >100 million rows dataset.For comparison, we will have two ways of getting counts of carriers.1. Serial way - Looping through each record and counting each airline's flight2. Map reduce way - map, reduce and sort, and collect way to counn the flights<jupyter_code>import os
import glob
import pandas as pd
import utils
import data_handler<jupyter_output><empty_output><jupyter_text>Data Preperation<jupyter_code>data_handler.download_dataset()<jupyter_output>data downloaded. you can skip this step or delete data folder to download again.<jupyter_text>Variables<jupyter_code># all files under data folder
file_list = sorted(glob.glob(os.path.join('data', '*.csv.bz2')))
carrier_counts = {}<jupyter_output><empty_output><jupyter_text>Serial WayHere, we are getting a list of files under the data folder. The serial way requires to loop through all of the records and update one by one.<jupyter_code>%%time
for ind, data_file in enumerate(file_list):
    # read current data
    df = pd.read_csv(data_file, encoding='ISO-8859-1', memory_map=True, low_memory=False)
    
    # unique airlines in dataset
    carriers = df.UniqueCarrier.unique()
    
    # update the global carrier_count
    for key in carriers:
        if key not in carrier_counts:
            carrier_counts.update({key: 0})
    
    # loop through each row in dataframe 
    for carrier in df.UniqueCarrier:
        carrier_counts[carrier] += 1

    # info 
    prefix ='Shape: {} ; {} Mb'.format(
        df.shape, round(df.memory_usage().sum() / 1e+6,2))
    
    # track the progress
    utils.progressbar(len(file_list), ind + 1, prefix=prefix)
# results
carrier_counts<jupyter_output><empty_output><jupyter_text>Map Reduce WaySo This dataset is very big have millions of records that's why i will use two columns 'FlightNum', 'UniqueCarrier' for creating a dataframe for running the MapReduce Algorithm.<jupyter_code>from functools import reduce
from itertools import groupby

data = pd.DataFrame()
cols_to_Load = ['FlightNum', 'UniqueCarrier']
for ind, data_file in enumerate(file_list):
    df = pd.DataFrame(pd.read_csv(data_file, encoding='ISO-8859-1', memory_map=True, low_memory=False, usecols= cols_to_Load))
    data=data.append(df)
carriers = data.UniqueCarrier
%%time
# phase 1.  Mapping

mapping=map(lambda x:(x,1),carriers) 
#print(list(mapping))
%%time
# Phase 2.  Shuffling or Sorting

sorted_mapping=sorted(mapping)
#print(sorted_mapping)
%%time
# Phase 3. Reducing
grouper=groupby(sorted_mapping, lambda p:p[0])
final=map(lambda l:(l[0],reduce(lambda x, y:x+y,map(lambda p: p[1], l[1]))), grouper)

print(list(final))<jupyter_output>[('9E', 521059), ('AA', 14984647), ('AQ', 154381), ('AS', 2878021), ('B6', 811341), ('CO', 8145788), ('DH', 693047), ('DL', 16547870), ('EA', 919785), ('EV', 1697172), ('F9', 336958), ('FL', 1265138), ('HA', 274265), ('HP', 3636682), ('ML (1)', 70622), ('MQ', 3954895), ('NW', 10292627), ('OH', 1464176), ('OO', 3090853), ('PA (1)', 316167), ('PI', 873957), ('PS', 83617), ('TW', 3757747), ('TZ', 208420), ('UA', 13299817), ('US', 14075530), ('WN', 15976022), ('XE', 2350309), ('YV', 854056)]
Wall time: 41 s
{'fraction_non_alphanumeric': 0.13561847988077497, 'fraction_numerical': 0.06885245901639345, 'mean_word_length': 5.2611940298507465, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '31482', 'license': 'MIT', 'n_tokens_mistral': 1297, 'n_tokens_neox': 1111, 'n_words': 347, 'path': 'Map_Reduce_Demonstration.ipynb', 'repo_name': 'Sakshi09wal/Map-Reduce-Algorithm', 'size': '3355'}