<s> import csv <EOL> from operator import add <EOL> from datetime import datetime <EOL> from StringIO import StringIO <EOL> from collections import namedtuple <EOL> from pyspark import SparkConf , SparkContext <EOL> DATE_FMT = "<STR_LIT>" <EOL> Order = namedtuple ( '<STR_LIT>' , ( '<STR_LIT:id>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:date>' ) ) <EOL> def parse ( row ) : <EOL> """<STR_LIT>""" <EOL> row [ <NUM_LIT:0> ] = int ( row [ <NUM_LIT:0> ] ) <EOL> row [ <NUM_LIT:2> ] = int ( row [ <NUM_LIT:2> ] ) <EOL> row [ <NUM_LIT:3> ] = datetime . strptime ( row [ <NUM_LIT:3> ] , DATE_FMT ) <EOL> return Order ( * row ) <EOL> def split ( line ) : <EOL> """<STR_LIT>""" <EOL> reader = csv . reader ( StringIO ( line ) ) <EOL> return reader . next ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> conf = SparkConf ( ) . setMaster ( "<STR_LIT>" ) . setAppName ( "<STR_LIT>" ) <EOL> sc = SparkContext ( conf = conf ) <EOL> orders = sc . textFile ( "<STR_LIT>" ) <EOL> orders = orders . map ( split ) . map ( parse ) <EOL> orders = orders . filter ( lambda order : order . date . year == <NUM_LIT> ) <EOL> orders . cache ( ) <EOL> months = orders . map ( lambda order : ( ( order . date . year , order . date . month ) , <NUM_LIT:1> ) ) <EOL> products = orders . filter ( lambda order : order . upc == "<STR_LIT>" ) <EOL> print products . toDebugString ( ) <EOL> months = months . reduceByKey ( add ) <EOL> print months . toDebugString ( ) <EOL> print months . take ( <NUM_LIT:5> ) <EOL> print products . count ( ) </s>
