For this project, we are leveraging Spark as the new engine to redesign the Hadoop auto post-sales report project that was previously done in MapReduce.

from pyspark import SparkContext
sc = SparkContext('local', 'autops')
raw_rdd = sc.textFile("data.csv") raw_rdd.take(2)

['1,I,VXIO456XLBB630221,Nissan,Altima,2003,2002-05-08,Initial sales from TechMotors', '2,I,INU45KIOOPA343980,Mercedes,C300,2015,2014-01-01,Sold from EuroMotors']

def extract_vin_key_value(line):
arr = line.split(',') return (arr[2], (arr[3], arr[5])) vin_kv = raw_rdd.map(lambda x: extract_vin_key_value(x)) vin_kv.collect()

[('VXIO456XLBB630221', ('Nissan', '2003')), ('INU45KIOOPA343980', ('Mercedes', '2015')), ('VXIO456XLBB630221', ('', '')), ('VXIO456XLBB630221', ('', '')), ('VOME254OOXW344325', ('Mercedes', '2015')), ('VOME254OOXW344325', ('', '')), ('VXIO456XLBB630221', ('', '')), ('EXOA00341AB123456', ('Mercedes', '2016')), ('VOME254OOXW344325', ('', '')), ('VOME254OOXW344325', ('', '')), ('EXOA00341AB123456', ('', '')), ('EXOA00341AB123456', ('', '')), ('VOME254OOXW344325', ('', '')), ('UXIA769ABCC447906', ('Toyota', '2017')), ('UXIA769ABCC447906', ('', '')), ('INU45KIOOPA343980', ('', ''))]

def populate_make(lines):
#first pass to find the make and year

for ln in lines:
    if ln[0] != '':
        make = ln[0]
        year = ln[1]
        break
#end pass to fill in make and year
new_lines = [(make, year) for ln in lines]

return new_lines

enhance_make = vin_kv.groupByKey().flatMap(lambda x: populate_make(list(x[1]))) enhance_make.collect()

[('Nissan', '2003'), ('Nissan', '2003'), ('Nissan', '2003'), ('Nissan', '2003'), ('Mercedes', '2015'), ('Mercedes', '2015'), ('Mercedes', '2015'), ('Mercedes', '2015'), ('Mercedes', '2015'), ('Mercedes', '2015'), ('Mercedes', '2015'), ('Mercedes', '2016'), ('Mercedes', '2016'), ('Mercedes', '2016'), ('Toyota', '2017'), ('Toyota', '2017')]

make_year_count = make_kv.reduceByKey(lambda a,b: a+b) make_year_count.collect()

[('Nissan-2003', 4), ('Mercedes-2015', 7), ('Mercedes-2016', 3), ('Toyota-2017', 2)]