Spark code

from pyspark.sql.types import *
from pyspark.sql.functions import *
import json, csv

from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext

#looks in the local directory for the AWS credentials
with open("credentials.csv") as f:
	access_key_id, secret_access_key = f.readlines()[1].split(',')[1:]

sc = SparkContext()

#this approach may not be optimal, as it's accessing a 'private' member of sc. The other approaches of modifying conf/spark-env.sh or including it in the s3 URL seem to have some security problems
sc._jsc.hadoopConfiguration().set("fs.s3n.awsAccessKeyId", access_key_id)
sc._jsc.hadoopConfiguration().set("fs.s3n.awsSecretAccessKey", secret_access_key)
sqlContext = SQLContext(sc)

#load users data
users = sc.textFile("s3n://ds12-education/clickstream-challenge/data/users.tsv.gz")
users_header = users.first()
users = users.filter(lambda elem: elem != users_header).map(lambda elem: elem.split('\t')).cache()

#load prodocuts data
products = sc.textFile("s3n://ds12-education/clickstream-challenge/data/products.tsv.gz")
products_header = products.first()
products = products.filter(lambda elem: elem != products_header).map(lambda elem: elem.split('\t')).cache()

#You can argue this is a little clunky, but its a good framework if we needed to do multiple, unique transformations on the clickstream columns. In the present case we are merely subsetting the columns and doing a transformation on a single column
def clickstream_columns(elem):
	output_columns = list()
	#we're going to subset some of the columns for clickstream
	columns = [1, 7, 12, 13, 49, 50, 52] #"timestamp ip_address url swid city country state"
	for column in columns:
		if column == 13: #additionally, clickstream.swid,c13, has beginning and ending braces not found in users.swid
			output_columns.append(elem[column].strip('{}'))
		else:
			output_columns.append(elem[column])
	return output_columns

#load click data
clickstream = sc.textFile("s3n://ds12-education/clickstream-challenge/data/Omniture*").map(lambda elem: elem.split('\t')).map(clickstream_columns).cache()

#exploratory data analysis
#users.count() #38,455
#products.count() #31
#clickstream.count() #421,266

#now we create DataFrames from the above RDDs
users_schemaString = "swid birth_dt gender_cd"
products_schemaString = "url category"
clickstream_schemaString = "timestamp ip_address url swid city country state"

def rdd_to_dataframe(rdd, schemaString):
	fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()]
	schema = StructType(fields)
	return sqlContext.createDataFrame(rdd, schema)

users_df = rdd_to_dataframe(users, users_schemaString)
products_df = rdd_to_dataframe(products, products_schemaString)
clickstream_df = rdd_to_dataframe(clickstream, clickstream_schemaString)

#to answer the questions of interest, we must join the datasets
click_products = clickstream_df.join(products_df, 'url').cache()
all_data = click_products.join(users_df, 'swid', how='left_outer').cache()

#distinct users and total number of clicks per day
groupedData = all_data.groupBy(year(all_data.timestamp), month(all_data.timestamp), dayofmonth(all_data.timestamp))
clicks_per_day = groupedData.count()
unique_users_per_day = groupedData.agg(countDistinct('swid'))
clicks_distint_users = unique_users_per_day.join(clicks_per_day, ['year(timestamp)', 'month(timestamp)', 'dayofmonth(timestamp)'])

#total number of clicks per product category
clicks_per_product = click_products.groupBy('category').count()

#generate TSV files
with open('clicks_per_product.tsv', 'wb') as tsvfile:
	tsvwriter = csv.writer(tsvfile, delimiter='\t')
	tsvwriter.writerow(['category', 'count'])
	tsvwriter.writerows(clicks_per_product.collect())

with open('clicks_distint_users.tsv', 'wb') as tsvfile:
	tsvwriter = csv.writer(tsvfile, delimiter='\t')
	tsvwriter.writerow(['year', 'month', 'day', 'distinct_users', 'clickcount'])
	tsvwriter.writerows(clicks_distint_users.collect())

#the following generates the JSON of interest; fundamentally we use the groupByKey method for PairRDDs to generate the collection of clicks for each user
nested_rdd = all_data.rdd.map(lambda elem: ((elem.swid, elem.birth_dt, elem.gender_cd), {'url': elem.url, 'timestamp': elem.timestamp, 'ipAddress': elem.ip_address, 'city': elem.city, 'state': elem.state, 'country': elem.country, 'category': elem.category})).groupByKey().map(lambda (key, value): json.dumps({'UserId': key[0], 'birthDate': key[1], 'genderCd': key[2], 'clicks': list(value)}))

#the default case is not to coalesce before writing the output, especially if the dataset is large. In this case, the instructions said to produce a "JSON file" and the data-size is plausible for a single file ;)
nested_rdd.coalesce(1).saveAsTextFile('json_single_output')