import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import datetime

import seaborn as sns
plt.style.use('seaborn-white')
%matplotlib inline

pro=pd.read_csv("datasets/promoted_content.csv")

# read page_views in pyspark
df_page_views = spark.read.csv(
    path        =   "datasets/page_views_sample.csv",
    header      =   True, 
    mode        =   "DROPMALFORMED",   # Poorly formed rows in CSV are dropped rather than erroring entire operation
    inferSchema =   True               # Not always perfect but works well in most cases as of 2.1+
)

df_page_views.printSchema()

df_page_views.show(n=5)

df_page_views.describe().show()

df_page_views.dtypes

df_page_views.select('timestamp').describe().show()

# Page view sample data is from 6/14 only
max_timestamp = (86399982 + 1465876799998)/1000.0
min_timestamp = (12 ++ 1465876799998)/1000.0

print 'start time:', datetime.datetime.fromtimestamp(min_timestamp).strftime('%Y-%m-%d %H:%M:%S.%f')
print 'end time:', datetime.datetime.fromtimestamp(max_timestamp).strftime('%Y-%m-%d %H:%M:%S.%f')

df_page_views.createOrReplaceTempView("temp_df_page_views")

# impressions by document_id
spark.sql('Create Table df_pv_by_doc AS (SELECT document_id, count(uuid) as impressions \
FROM temp_df_page_views Group by document_id Order by impressions DESC)').show(5)

spark.sql('Select * from df_pv_by_doc').show(5)

spark.sql('Select avg(impressions) FROM df_pv_by_doc').show()

# impressions by user 
spark.sql('Create Table df_pv_by_uuid AS (SELECT uuid, count(document_id) as impressions, \
    platform FROM temp_df_page_views Group by uuid, platform Order by impressions DESC)').show(5)

spark.sql('Select * from df_pv_by_uuid').show(5)

spark.sql('Select avg(impressions), platform FROM df_pv_by_uuid group by platform').show()

spark.sql('Select avg(impressions)FROM df_pv_by_uuid').show()

# Does an user view same document more than once within same day? - No
spark.sql('SELECT uuid, document_id, count(timestamp) FROM temp_df_page_views Group by uuid, document_id Order by count(timestamp) DESC').show(5)

df_page_views_pd = pd.read_csv("datasets/page_views_sample.csv")

# engagement level (pageviews per day by user)
page_views_per_user = df_page_views_pd.groupby(['uuid'])['document_id'].count()

# documents popularity (average daily impressions)
page_views_per_doc = df_page_views_pd.groupby(['document_id'])['uuid'].count()

# read events in pyspark
df_events = spark.read.csv(
    path        =   "datasets/events.csv",
    header      =   True, 
    mode        =   "DROPMALFORMED",   # Poorly formed rows in CSV are dropped rather than erroring entire operation
    inferSchema =   True               # Not always perfect but works well in most cases as of 2.1+
)

df_events.show(n=5)

df_events.select('timestamp').describe().show()

df_events.dtypes

df_events_pd = pd.read_csv("datasets/events.csv")

# events data is from 6/14 - 6/28
max_timestamp = (1295999805 + 1465876799998)/1000.0
min_timestamp = (61 ++ 1465876799998)/1000.0

print 'start time:', datetime.datetime.fromtimestamp(min_timestamp).strftime('%Y-%m-%d %H:%M:%S.%f')
print 'end time:', datetime.datetime.fromtimestamp(max_timestamp).strftime('%Y-%m-%d %H:%M:%S.%f')

# What time do people view ads most (by hour)
df_events_pd['datetime'] = df_events_pd.timestamp.apply(lambda x: datetime.datetime.fromtimestamp((x+1465876799998)/1000.0).\
                             strftime('%Y-%m-%d %H:%M:%S.%f'))

df_events_pd['hour'] = df_events_pd['datetime'].apply(lambda x: x[11:13])

df_events_pd['day'] = df_events_pd['datetime'].apply(lambda x: x[:10])

hour_table = df_events_pd.groupby(['hour'])['display_id'].count()

day_table = df_events_pd.groupby(['day'])['display_id'].count()

objects = hour_table.index
y_pos = np.arange(len(objects))

fig, ax = plt.subplots(figsize=(16,7))
plt.bar(y_pos, hour_table, align='center', alpha=0.5)
plt.xticks(y_pos, objects)
plt.ylabel('Ads impressions')
plt.title('Number of ads viewed by hour', fontsize=14);

# What date do people viewed ads most (by day)
objects = day_table.index
y_pos = np.arange(len(objects))

fig, ax = plt.subplots(figsize=(16,7))
plt.bar(y_pos, day_table, align='center', alpha=0.5)
plt.xticks(y_pos, objects)
plt.ylabel('Ads impressions')
plt.title('Number of ads viewed by date', fontsize=14);

day_df_table = pd.DataFrame(day_table)
day_df_table.index = pd.DatetimeIndex(day_df_table.index)

day_df_table['dayofweek']=day_df_table.index.dayofweek

dayofweek_df_table = day_df_table.groupby(['dayofweek']).mean()

# What day of week do people viewed ads most (by day)
objects = ['Mon','Tue','Wed','Thu','Fri','Sat','Sun']
y_pos = np.arange(len(objects))

fig, ax = plt.subplots(figsize=(12,7))
plt.bar(y_pos, dayofweek_df_table['display_id'], align='center', alpha=0.5)
plt.xticks(y_pos, objects)
plt.ylabel('Ads impressions')
plt.title('Number of ads viewed by day', fontsize=14);

df_events.createOrReplaceTempView("temp_df_events")

# number of display per document
spark.sql('Create table df_events_per_doc AS (SELECT document_id, count(display_id) as ad_impr \
        FROM temp_df_events Group by document_id Order by ad_impr DESC)').show(5)

spark.sql('Select * from df_events_per_doc').show(5)

spark.sql('Select avg(ad_impr) from df_events_per_doc').show(5)

# % of documents viewed
df_events_pd_document_table = df_events_pd.groupby(['document_id']).count().groupby(['display_id']).count()

total_documents = df_events_pd_document_table.uuid.sum()
df_events_pd_document_table_percentage = df_events_pd_document_table.uuid.apply(lambda x: x/float(total_documents)*100)

print '-- {:.2f}% documents were viewed once with ads'.format(df_events_pd_document_table_percentage.iloc[0])
for i in [5, 10, 20, 100,1000,10000]:
    print '-- {:.2f}% documents were viewed more than {} times with ads'.format(df_events_pd_document_table_percentage.loc[i:].sum(),i)

# number of display viewed per user (which platform)
spark.sql('Create table df_events_per_uuid AS (SELECT uuid, count(display_id) as ad_impr, \
platform FROM temp_df_events Group by uuid, platform Order by ad_impr DESC)').show(5)

spark.sql('Select * from df_events_per_uuid').show(5)

spark.sql('Select avg(ad_impr) from df_events_per_uuid').show(5)

spark.sql('Select avg(ad_impr),platform from df_events_per_uuid group by platform').show(5)

# % of users who viewed ads more than once
df_events_pd_user_table = df_events_pd.groupby(['uuid']).count().groupby(['display_id']).count()

total_users = df_events_pd_user_table.document_id.sum()
df_events_pd_user_table_percentage = df_events_pd_user_table.document_id.apply(lambda x: x/float(total_users)*100)

print '-- {:.2f}% users viewed ads only once'.format(df_events_pd_user_table_percentage.iloc[0])
print '-- {:.2f}% users viewed ads twice'.format(df_events_pd_user_table_percentage.iloc[1])
print '-- {:.2f}% users viewed ads three times'.format(df_events_pd_user_table_percentage.iloc[2])
print '-- {:.2f}% users viewed ads four times'.format(df_events_pd_user_table_percentage.iloc[3])
print '-- {:.2f}% users viewed ads more than five times'.format(df_events_pd_user_table_percentage.iloc[4:].sum())

df_events_pd.platform.unique()

df_events_pd.platform = df_events_pd.platform.astype(str)

# plot user by platform
df_events_platform_table = df_events_pd.pivot_table(index='platform',values='uuid', aggfunc=[len])

# 1 = desktop, 2 = mobile, 3 = tablet
objects = ('desktop', 'mobile', 'tablet', 'NA')
y_pos = np.arange(len(objects))

fig, ax = plt.subplots(figsize=(12,7))
plt.bar(y_pos, df_events_platform_table.len, align='center', alpha=0.5)
plt.xticks(y_pos, objects)
plt.ylabel('Number of users')
plt.title('Number of users by platform', fontsize=14)

rects = ax.patches
values = df_events_platform_table.len.tolist()

for rect, value in zip(rects, values):
    height = rect.get_height()
    ax.text(rect.get_x() + rect.get_width()/2, height+5, value, ha='center', va='bottom')

platform_list =['desktop','mobile','tablet']
for i in range(3):
    platform = df_events_platform_table.len[i] / float(df_events_platform_table.len.sum()) * 100
    print '-- {:.2f}% users came from {}'.format(platform,platform_list[i]) 

# relationship between events and page_views on 6/14
# left join page_view sample and events on uuid, document, time

spark.sql('Create Table df_events_pv AS (SELECT e.display_id, pv.uuid, pv.document_id, \
    pv.timestamp, pv.platform, pv.geo_location, \
    pv.traffic_source FROM temp_df_page_views as pv left join temp_df_events as e \
    on pv.uuid = e.uuid and pv.document_id = e.document_id)').show(5)

# first 5 rows from the new joined table
spark.sql('select * from df_events_pv limit 5').show()

# display rate on 6/14 - 1.29% (number of display_id / number of uuid)
spark.sql('SELECT count(display_id) / count(uuid) from df_events_pv').show()

# display rate on 6/14 by platform - 3 (tablet) has the highest display rate, 1 (desktop) is
# the second, 2 (mobile) is the last one
spark.sql('SELECT count(display_id) / count(uuid), platform from df_events_pv \
        group by platform').show()

df_events_pd['country'] = df_events_pd.geo_location.astype('str').apply(lambda x: x[0:2])

def extract_state(x):
    if str(x)[0:2] == 'US':
        try:
            y = str(x)[3:5]
        except:
            y = np.nan
    else:
        y = 'Outside US'
    return y

df_events_pd['state'] = df_events_pd.geo_location.apply(lambda x: extract_state(x))

df_location_table = df_events_pd.groupby(['state'])['document_id'].count()

df_location_table =df_location_table.sort_values(ascending=False)

df_location_table = pd.DataFrame(df_location_table)
df_location_table

total_item = float(df_location_table.document_id.sum())

df_location_table['percentage'] = df_location_table.document_id.map(lambda x: x/total_item*100)

df_location_table.head()

fig,ax = plt.subplots(figsize=(15, 7))
sns.barplot(df_location_table.index, df_location_table.values, order=df_location_table.index, alpha=1)
plt.ylabel('Total impressions')
plt.title('Ad Impression by location', fontsize=14);

# read categories in pyspark
df_categories = spark.read.csv(
    path        =   "datasets/documents_categories.csv",
    header      =   True, 
    mode        =   "DROPMALFORMED",   # Poorly formed rows in CSV are dropped rather than erroring entire operation
    inferSchema =   True               # Not always perfect but works well in most cases as of 2.1+
)

# read entities in pyspark
df_entities = spark.read.csv(
    path        =   "datasets/documents_entities.csv",
    header      =   True, 
    mode        =   "DROPMALFORMED",   # Poorly formed rows in CSV are dropped rather than erroring entire operation
    inferSchema =   True               # Not always perfect but works well in most cases as of 2.1+
)

# read meta in pyspark
df_meta = spark.read.csv(
    path        =   "datasets/documents_meta.csv",
    header      =   True, 
    mode        =   "DROPMALFORMED",   # Poorly formed rows in CSV are dropped rather than erroring entire operation
    inferSchema =   True               # Not always perfect but works well in most cases as of 2.1+
)

# read topics in pyspark
df_topics = spark.read.csv(
    path        =   "datasets/documents_topics.csv",
    header      =   True, 
    mode        =   "DROPMALFORMED",   # Poorly formed rows in CSV are dropped rather than erroring entire operation
    inferSchema =   True               # Not always perfect but works well in most cases as of 2.1+
)

df_categories_pd = pd.read_csv("datasets/documents_categories.csv")

df_entities_pd = pd.read_csv("datasets/documents_entities.csv")

df_meta_pd = pd.read_csv("datasets/documents_meta.csv")

df_topics_pd = pd.read_csv("datasets/documents_topics.csv")

df_categories_pd.head()

len(df_categories_pd.category_id.unique())

categories_table = df_categories_pd.groupby(['category_id'])['document_id'].count()

categories_table = categories_table.sort_values()

categories_table[categories_table>1000].sum()

fig,ax = plt.subplots(figsize=(12, 7))
sns.barplot(categories_table.index, categories_table.values, order=categories_table.index, alpha=1)
plt.ylabel('Total occurences')
plt.title('Category occurences', fontsize=14);

total_num = float(len(df_categories_pd.category_id.unique()))
for i in [10, 1000, 10000, 50000, 100000,500000]:
    print '-- {:.3f}% of categories appeared in more than {} documents'.format(categories_table[categories_table>i].count() / total_num *100,i)

df_topics_pd.head()

len(df_topics_pd.topic_id.unique())

topics_table = df_topics_pd.groupby(['topic_id'])['document_id'].count()

topics_table = topics_table.sort_values()

topics_table

total_num_2 = float(300)
for i in [5000, 10000, 50000,100000,200000]:
    print '-- {:.3f}% of topics appeared in more than {} documents'.format(topics_table[topics_table>i].count() / total_num_2 *100,i)
    
    
fig,ax = plt.subplots(figsize=(12, 7))
sns.barplot(topics_table.index, topics_table.values, order=topics_table.index, alpha=1)
plt.ylabel('Total occurences')
plt.title('Topic occurences', fontsize=14);

bin = np.arange(0,1.1,0.1)
group_name = np.arange(1,11)
df_categories_pd['c_level'] = pd.cut(df_categories_pd.confidence_level,bin,labels=group_name)
df_topics_pd['c_level'] = pd.cut(df_topics_pd.confidence_level,bin,labels=group_name)

# read content in pyspark
df_content = spark.read.csv(
    path        =   "datasets/promoted_content.csv",
    header      =   True, 
    mode        =   "DROPMALFORMED",   # Poorly formed rows in CSV are dropped rather than erroring entire operation
    inferSchema =   True               # Not always perfect but works well in most cases as of 2.1+
)

# read content in pyspark
df_train = spark.read.csv(
    path        =   "datasets/clicks_train.csv",
    header      =   True, 
    mode        =   "DROPMALFORMED",   # Poorly formed rows in CSV are dropped rather than erroring entire operation
    inferSchema =   True               # Not always perfect but works well in most cases as of 2.1+
)

# read content in pyspark
df_test = spark.read.csv(
    path        =   "datasets/clicks_test.csv",
    header      =   True, 
    mode        =   "DROPMALFORMED",   # Poorly formed rows in CSV are dropped rather than erroring entire operation
    inferSchema =   True               # Not always perfect but works well in most cases as of 2.1+
)

df_train_pd = pd.read_csv("datasets/clicks_train.csv")

df_test_pd = pd.read_csv("datasets/clicks_test.csv")

df_test_pd.head()

# Ads occurences
df_train_table = df_train_pd.groupby(['ad_id'])['display_id'].count()

df_train_table.sort_values()

total_ads = float(len(df_train_table))
print '-- {:.2f}% ads appeared once'.format(len(df_train_table[df_train_table==1])/total_ads*100)
for i in [5,10, 100, 1000, 10000, 100000]:
    print '-- {:.2f}% ads appeared more than {} times'.format(len(df_train_table[df_train_table>i])/total_ads*100, i)

# % of test ads in train data
df_test_table = df_test_pd.groupby(['ad_id'])['display_id'].count()

result = (len(df_test_table.index[df_test_table.index.isin(df_train_table.index)])/float(len(df_test_table.index)))

print '{:.2f}% ads in test dataset appeared in training dataset'.format(result*100)