Apache Spark - Analyzing Food Insecurity in NYC using KeyFoods Price Catalogs¶

%%shell
gdown --quiet 1O1U_t-cpmValVK2mjdTzcFxIbGw05vOw
gdown --quiet 1YUBKrtNV3QUz1RutMnMbJdQj7rv-Lkd5
gdown --quiet 1f79oETtvN3NQLYPnVGhurE1UBDP4IQP-
pip install pyspark

import csv
import json
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
import IPython
%matplotlib inline
IPython.display.set_matplotlib_formats('svg')
pd.plotting.register_matplotlib_converters()
sns.set_style("whitegrid")

import pyspark
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql import types as T
sc = pyspark.SparkContext.getOrCreate()
spark = SparkSession(sc)
spark

#header's for products
products = sc.textFile('keyfood_products.csv',use_unicode=True)

list(enumerate(products.first().split(',')))

#headers for sample items 
sample_items = sc.textFile('keyfood_sample_items.csv')

list(enumerate(sample_items.first().split(',')))

#loading products, getting the UPC, store, price, and doing manipulations
#like getting second part of UPC code, converting price to float 
products = sc.textFile('keyfood_products.csv',use_unicode=True).mapPartitionsWithIndex(extract_products_data)\
  .map(lambda x: (x[2].split('-')[1],(x[0],float(x[5].split(u'\xa0')[0].replace('$','')))))

print(products.count())
products.take(5)

#loading in samples into rdd, and splitting UPC and getting the second part 
sample_items = sc.textFile('keyfood_sample_items.csv').mapPartitionsWithIndex(extract_items_data)\
  .map(lambda x: (x[0].split('-')[1],(x[1],)))

print(sample_items.count())
sample_items.take(5)

#import lib for reading json
import json

#define generator function for extracting data from keyfood_nyc_stores.json
def extract_json_data(json_f):
  #opening file 
  with open(json_f) as f:
    #loading json file using json.load
    data = json.load(f)
    #looping through all the keys
    for key in data.keys():
      #for each record, extract the needed info 
      record = data[key]
      yield (record['name'],(record['communityDistrict'],record['foodInsecurity']))

#convert into json into rdd
nyc_stores = sc.parallelize(extract_json_data('keyfood_nyc_stores.json'))

nyc_stores.take(4)

#joining the sample_items, products, and nyc_stores rdd to get desired output 
outputTask1 = sample_items.join(products).mapValues(lambda x: x[0]+x[1])\
  .map(lambda x: (x[1][1],(x[1][0],x[1][2]))).join(nyc_stores.mapValues(lambda x: (int(round(x[1]*100)),)))\
  .mapValues(lambda x: x[0]+x[1]).values()

## DO NOT EDIT BELOW
outputTask1 = outputTask1.cache()
outputTask1.count()

outputTask1.take(5)

#@title
def dfTask1(data):
    rdd = data.rdd if hasattr(data, 'rdd') else data
    if rdd.count()>10000:
        raise Exception('`outputTask1` has too many rows')
    rows = map(lambda x: (x[0], x[1], int(x[2])), rdd.collect())
    return pd.DataFrame(data=rows, columns=['Item Name','Price ($)','% Food Insecurity'])

def plotTask1(data, figsize=(8,8)):
    itemNames = pd.read_csv('keyfood_sample_items.csv')['Item Name']
    itemKey = dict(map(reversed,enumerate(itemNames)))
    df = dfTask1(data).sort_values(
        by = ['Item Name', '% Food Insecurity'],
        key = lambda x: list(map(lambda y: itemKey.get(y,y), x)))
    plt.figure(figsize=figsize)
    ax = sns.violinplot(x="Price ($)", y="Item Name", data=df, linewidth=0,
                        color='#ddd', scale='width', width=0.95)
    idx = len(ax.collections)
    sns.scatterplot(x="Price ($)", y="Item Name", hue='% Food Insecurity', data=df,
                    s=24, linewidth=0.5, edgecolor='gray', palette='YlOrRd')
    for h in ax.legend_.legendHandles: 
        h.set_edgecolor('gray')
    pts = ax.collections[idx]
    pts.set_offsets(pts.get_offsets() + np.c_[np.zeros(len(df)),
                                            np.random.uniform(-.1, .1, len(df))])
    ax.set_xlim(left=0)
    ax.xaxis.grid(color='#eee')
    ax.yaxis.grid(color='#999')
    ax.set_title('Item Prices across KeyFood Stores in NYC')
    return ax

if 'outputTask1' not in locals():
    raise Exception('There is no `outputTask1` produced in Task 1.A')

plotTask1(outputTask1);