<jupyter_start><jupyter_text>Final Project: Click-through Rate Prediction__`MIDS w261: Machine Learning at Scale | UC Berkeley School of Information | Spring 2019`__We've been provided a week’s worth of Criteo display advertising data to develop models predicting ad click-through rate (CTR). Our aim is to generate precise and accurate probabilities that a consumer will click on a given ad based on the user and the page visited. About the Data:The dataset contains 2 files: train.txt and test.txt. The training dataset consists of a portion of Criteo's traffic over a period of 7 days. Each row corresponds to a display ad served by Criteo and the first column indicates whether this ad has been clicked or not. The positive (clicked) and negatives (non-clicked) examples have both been subsampled (but at different rates) in order to reduce the dataset size. The rows are chronologically ordered.There are 13 features taking integer values (mostly count features) and 26 categorical features. The values of the categorical features have been hashed onto 32 bits for anonymization purposes. The semantic of these features is undisclosed. Some features may have missing values. The test set is computed in the same way as the training set but it corresponds to events on the day following the training period. The first column (label) has been removed. Rubric under Consideration:1. Question Formulation -- Introduce the goal of your analysis. What questions will you seek to answer, why do people perform this kind of analysis on this kind of data? Preview what level of performance your model would need to achieve to be practically useful.2. Algorithm Explanation -- Create your own toy example that matches the dataset provided and use this toy example to explain the math behind the algorithm that you will perform.3. EDA & Discussion of Challenges -- Determine 2-3 relevant EDA tasks that will help you make decisions about how you implement the algorithm to be scalable. Discuss any challenges that you anticipate based on the EDA you perform.4. Algorithm Implementation -- develop a ‘homegrown’ implementation of the algorithm, apply it to the training dataset and evaluate your results on the test set.5. Application of Course Concepts -- Pick 3-5 key course concepts and discuss how your work on this assignment illustrates an understanding of these concepts.  Notebook SetupBefore starting, run the following cells to confirm your setup.<jupyter_code># imports
import re
import ast
import time
import itertools
import numpy as np
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt
%matplotlib inline
%reload_ext autoreload
%autoreload 2
# store path to notebook
PWD = !pwd
PWD = PWD[0]
# start Spark Session
from pyspark.sql import SparkSession
app_name = "final"
master = "local[*]"
spark = SparkSession\
        .builder\
        .appName(app_name)\
        .master(master)\
        .getOrCreate()
sc = spark.sparkContext
#pyspark dependencies
from pyspark.ml import Pipeline
from pyspark.ml.feature import OneHotEncoderEstimator, StringIndexer, VectorAssembler
from pyspark.sql import SQLContext, functions as f
from pyspark.sql.types import *
# make a data folder, download, and unpack file
!mkdir data
!wget "https://s3-eu-west-1.amazonaws.com/kaggle-display-advertising-challenge-dataset/dac.tar.gz"
!tar xvzf dac.tar.gz
!mv *.txt data<jupyter_output><empty_output><jupyter_text>Looking into Test Data<jupyter_code>#----per readme: when a value is missing, the field is just empty---#
file = "data/test.txt"
df = spark.read.schema(schema) \
    .option("delimiter", "\t") \
    .option("ignoreLeadingWhiteSpace",True) \
    .option("ignoreTrailingWhiteSpace",True) \
    .csv(file,nullValue = "")
intCols = [i for i in df.columns if "int" in i]
df.select(intCols).describe().write \
    .option("header", "true") \
    .csv("eda/tester")
#df.columns

 ['intFeature1',
 'intFeature2',
 'intFeature3',
 'intFeature4',
 'intFeature5',
 'intFeature6',
 'intFeature7',
 'intFeature8',
 'intFeature9',
 'intFeature10',
 'intFeature11',
 'intFeature12',
 'intFeature13']
#consolidating summary files
import glob
files = np.sort(glob.glob('eda/test/*.csv'))
summary = pd.concat([pd.read_csv(f) for f in files],ignore_index = True)
summary
#consolidating summary files
import glob
files = np.sort(glob.glob('data/Deduped_initSummary.csv/*.csv'))
train_summary = pd.concat([pd.read_csv(f) for f in files],ignore_index = True)
train_summary[ ['summary','intFeature1',
 'intFeature2',
 'intFeature3',
 'intFeature4',
 'intFeature5',
 'intFeature6',
 'intFeature7',
 'intFeature8',
 'intFeature9',
 'intFeature10',
 'intFeature11',
 'intFeature12',
 'intFeature13']]<jupyter_output><empty_output><jupyter_text>Reading and Processing Data<jupyter_code>#set dataframe schema
#--- To avoid going through the entire data once, disable inferSchema option or specify the schema explicitly using schema

# 13 features taking integer values and 26 taking categorical values (we can't assume ordinal) - making total of 39 features + 1 outcome
#outcomeField = [StructField("label", IntegerType(), True)]
#outcomeField +
quantFields = [StructField("intFeature"+str(i+1), IntegerType(), True) for i in np.arange(13)]
qualFields = [StructField("catFeature"+str(i+1), StringType(), True) for i in np.arange(26)]
schema = StructType(quantFields + qualFields)
#read in the txt file
#----per readme: when a value is missing, the field is just empty---#
file = "data/train.txt"
df = spark.read.schema(schema) \
    .option("delimiter", "\t") \
    .option("ignoreLeadingWhiteSpace",True) \
    .option("ignoreTrailingWhiteSpace",True) \
    .csv(file,nullValue = "")
df.show(5)
#total number of rows in the dataset
df.count()
#checking for duplicate entries
dedupedDF = df.dropDuplicates()
dedupedDF.count()
print(str(45840617-45829966)+" duplicates removed from dataset")
#remove all rows that have missing values for each feature
deduped_NAdropped_DF = dedupedDF.dropna(how='all')
deduped_NAdropped_DF.count()
#check num of rows missing 3/4 or more of total features values
deduped_NAdropped_thresh = deduped_NAdropped_DF.dropna(thresh=30)
deduped_NAdropped_thresh.count()
print(str(45829966-45829966)+" rows had all of their data marked as null")
print(str(45829966-41879511)+" rows had 3/4 of their data marked as null")
#quick summary of the data
dfSummary = dedupedDF.describe()

dfSummary.write \
    .option("header", "true") \
    .csv("data/Deduped_initSummary.csv")
#consolidating summary files
import glob
files = np.sort(glob.glob('data/Deduped_initSummary.csv/*.csv'))
summary = pd.concat([pd.read_csv(f) for f in files],ignore_index = True)
#let's get a sense of data completeness
'''the 1st row from the summary table is count (excludes nulls), 
so we can divide that by the total rows for feature completeness'''
perc_complete = summary.iloc[0][1:].apply(lambda x: x/nrows)
#features with more than 10% of their values missing
perc_complete[perc_complete<.9].sort_values()
#summary of our integer variables
'''integer features are 1st 15 coumns in dataset'''
summary[summary.columns[0:15]].round(decimals = 2)<jupyter_output><empty_output><jupyter_text>~ 1/5 of our features are missing close to half or more values Initial Observations of Quantitative Features:* The ranges vary substantially across the integer features so we'll need to normalize these features to scale* Major variances between the max values and the mean values are suggestive of outliers and/or features with skewed distributions* intFeature2 has at least one negative value which seems erroneous  EDA of our Quant Features* will need to take a sample - discuss sample size decision (at least 50 observations per subset in categorical features???<jupyter_code>#create separate directory for EDA files
!mkdir eda
#approximate count of unique values per feature
from pyspark.sql.functions import col, approx_count_distinct

uniqueVals = deduped_NAdropped_DF \
                .agg(*(approx_count_distinct(col(c)).alias(c) for c in deduped_NAdropped_DF.columns)) \
                .collect()<jupyter_output><empty_output><jupyter_text>[See reference for counting unique values in each column of a Pyspark Dataframe](https://stackoverflow.com/questions/40888946/spark-dataframe-count-distinct-values-of-every-column)<jupyter_code>#converting to pandas dataframe for familiarity
uniqueCounts = pd.DataFrame(uniqueVals,columns = deduped_NAdropped_DF.columns) \
                .transpose() \
                .reset_index() \
                .sort_values(by = 0, ascending = False)


#write to eda directory for later use
fpath = "eda/featureCounts_unique.csv"
uniqueCounts.to_csv(fpath, header = ["Feature Name","Approximate Unique Values"], index = False)
#take a subset of the (deduped) training data for EDA - 5% in this case - and write to file for later use
trainSample = deduped_NAdropped_DF.sample(False, 0.05, 123)

trainSample.write \
    .option("header", "true") \
    .csv("eda/trainSample")
#load the trainSample into memory with a pandas dataframe
import glob
files = np.sort(glob.glob('eda/trainSample/*.csv'))
trainSample = pd.concat([pd.read_csv(f) for f in files],ignore_index = True)
trainSample.head()
# take a look at histograms for each integer feature
intFeatures = trainSample.columns[1:14]
trainSample[intFeatures].hist(figsize=(14,14), bins=100)
plt.show()<jupyter_output><empty_output><jupyter_text>we can see that our variables are heavily positive (right) skewed toward smaller integer values. We may be able to approximate normal distributions via log transformations. If we can demonstate approximate normal distributions among our variables, we can apply inferences from the logistic regression model. The logistic regression model assumes that the log-odds of an observation y can be expressed as a linear function of the input variables x.<jupyter_code>#checking how many negative values intFeature2 has
f2 = trainSample['intFeature2']
len(f2[f2<0])<jupyter_output><empty_output><jupyter_text>the number of negative entries, 236,565, in intFeature2 make up ~0.5% of all entries so it may be reasonable to assume that these should have been coded as missing values or may have been incorrectly assigned a negative value.<jupyter_code>#replace negative values with NaNs in intFeature2
neg_vals = f2[f2<0].unique()
trainSample.intFeature2 = trainSample['intFeature2'].replace(to_replace = neg_vals, value = np.NaN)
#helper function to log transform int features (should log transform first)
def log_transform(x):
    #takes a series
    '''shifts x up by one to account for presence of 0 values'
    and assumes negative values should be NaN'''
    return np.log(x+1)

#plotting log transformed integerFeatures
trainSample[intFeatures].apply(log_transform).hist(figsize=(14,14), bins=100)
plt.show()<jupyter_output><empty_output><jupyter_text>Even after log transformation the integer data is still very skewed, but it looks more manageable<jupyter_code>#apply log transformation to integer variables
trainSample[intFeatures] = trainSample[intFeatures].apply(log_transform)
#normalize log-transformed features to rescale
def normalize(series):
    mn = series.mean()
    std = series.std()
    
    return (series-mn)/std

#don't normalize label
trainSample[intFeatures] = trainSample[intFeatures].apply(normalize)
trainSample.head(5)
#write to file for later use
fpath = 'eda/trainSampleNorm.csv'
trainSample.to_csv(path_or_buf = fpath, index = False)
#read in normalized train data
trainSampleLogNorm = pd.read_csv('eda/trainSampleNorm.csv')
trainSampleLogNorm.head(5)
# plot heatmap for correlations matrix among integer features
intFeatures = trainSampleNorm.columns[1:14]

corr = trainSampleLogNorm[intFeatures].corr()
fig, ax = plt.subplots(figsize=(11, 9))
mask = np.zeros_like(corr, dtype=np.bool)
mask[np.triu_indices_from(mask)] = True
cmap = sns.diverging_palette(240, 10, as_cmap=True)
sns.heatmap(corr, mask=mask, cmap=cmap, center=0, linewidths=.5)
plt.title("Correlations between Integer features.")
plt.show()<jupyter_output><empty_output><jupyter_text>assumption in logistic regression is no perfect collinearity so not to worried about the few high correlations between:* intFeature 4 and intFeature13* intFeature 8 and intFeature13* intFeature 7 and intFeature11<jupyter_code>import seaborn as sns
# fig, ax_grid = plt.subplots(2, 1, figsize=(15,15)
sns.regplot(x='intFeature7', y="intFeature11",data=trainSampleLogNorm)                            
plt.show()
import seaborn as sns
# fig, ax_grid = plt.subplots(2, 1, figsize=(15,15)
sns.regplot(x='intFeature5', y="intFeature10",data=trainSampleLogNorm)                            
plt.show()<jupyter_output>/opt/anaconda/lib/python3.6/site-packages/scipy/stats/stats.py:1713: FutureWarning: Using a non-tuple sequence for multidimensional indexing is deprecated; use `arr[tuple(seq)]` instead of `arr[seq]`. In the future this will be interpreted as an array index, `arr[np.array(seq)]`, which will result either in an error or a different result.
  return np.add.reduce(sorted[indexer] * weights, axis=axis) / sumval<jupyter_text>Looking at the above scatter plot between intFeature7 and intFeature11 it appears that those two variables have close to perfect collinearity. This behavior can confuse our learning model....elaborate on this. May make sense to drop either intFeature11 or intFeature7<jupyter_code>#multivariate analysis against target feature
"""plot boxplots of each feature vs. the outcome"""
fig, ax_grid = plt.subplots(4, 4, figsize=(15,15))
y = trainSampleLogNorm['label']
for idx, feature in enumerate(intFeatures):
    x = trainSampleLogNorm[feature]
    sns.boxplot(x, y, ax=ax_grid[idx//4][idx%4], orient='h', linewidth=.5)
#     ax_grid[idx//3][idx%3].invert_yaxis()
fig.suptitle("Integer Features vs. Outcome(Click, No Click)", fontsize=15, y=0.9)
plt.show()<jupyter_output><empty_output><jupyter_text>outliers skewing data, so it's difficult to gauge differences in boxplots. may make sense to bin our integer datasee here for reference on [adaptive binning](https://towardsdatascience.com/understanding-feature-engineering-part-1-continuous-numeric-data-da4e47099a7b)transforming integer counts to interval data...maybe?so far, most integer features don't seem to show much linear association with click success or failure  EDA Categorical Data<jupyter_code>uniqueCounts = pd.read_csv('eda/featureCounts_unique.csv')
categoryLevels = uniqueCounts[uniqueCounts['Feature Name'].str.contains('cat')]
sns.barplot('Feature Name', 'Approximate Unique Values',data = categoryLevels)
plt.xticks(rotation=45)
plt.show()<jupyter_output><empty_output><jupyter_text>There are an extremely high number of levels within most categorical features and unfortunately we're given no information about the semantics of the levels within each category it's difficult to reduce these lacking that context. For instance, catFeature3 may be page content that could be reduced to a broader classification of content such as sports, finance, etc<jupyter_code>categoryLevels
#let's look at relative frequency of each level among our categories
def levelRarity(DataFrame,FeatureSet,threshold = 10):
    '''takes a dataset and given list of its categorical features and returns 
    the percentage of levels in a category with fewer observations than
    the threshold (rarity)'''
    #construct dictionary with features as keys
    fDict = {f:None for f in FeatureSet}

    for f in FeatureSet:
        x = DataFrame[f].value_counts()
        num = len(x[x<threshold])
        denom = len(x)
        #update feature value with percent less than threshold
        fDict[f] = num/denom

    return pd.DataFrame.from_dict(fDict, orient='index',columns = ["% of levels w/ obs < "+ str(threshold)])

FeatureSet = np.array(categoryLevels['Feature Name'])

levelRarity(trainSampleLogNorm,FeatureSet)
#iteratively build dataframe with categorical adjustments

for f in catFeatures:
    newCol_name = f + "_binned"
    row_elements = []
    #counts of unique labels per category
    uniq = trainSampleLogNorm[f].value_counts()
    #indx of labels with fewer than 30 observations (return the labels with .index)
    idx = set(uniq[uniq<30].index)
    labels = trainSampleLogNorm[f].values
    for label in labels:
        if label in idx:
            row_elements.append(f+"-bucket")
        else:
            row_elements.append(label)
            
    trainSampleLogNorm[newCol_name] = row_elements

#write to file for later use
fpath = 'eda/trainSampleLogNorm_Binned.csv'
trainSampleLogNorm.to_csv(path_or_buf = fpath, index = False)
#read in normalized train data
trainLogNormBinned = pd.read_csv('eda/trainSampleLogNorm_Binned.csv')
trainLogNormBinned.head(5)
#cleaning for train/test
#drop intFeature12 and catFeature22 based on columnwise incompleteness
trainLogNormBinned = trainLogNormBinned.drop(columns = ['intFeature12','catFeature22'])
#dropping rowise incompleteness (must be 3/4 complete)

#drop form cat variables (unbinned)
# unbinned_catFeatures = ['catFeature1', 'catFeature2', 'catFeature3', 'catFeature4',
#        'catFeature5', 'catFeature6', 'catFeature7', 'catFeature8',
#        'catFeature9', 'catFeature10', 'catFeature11', 'catFeature12',
#        'catFeature13', 'catFeature14', 'catFeature15', 'catFeature16',
#        'catFeature17', 'catFeature18', 'catFeature19', 'catFeature20',
#        'catFeature21', 'catFeature23', 'catFeature24', 'catFeature25',
#        'catFeature26']
# trainLogNormBinned = trainLogNormBinned.drop(columns = unbinned_catFeatures)

# trainLogNormBinned = trainLogNormBinned.drop(columns = ['catFeature22_binned'])
trainLogNormBinned.shape
trainLogNormBinned = trainLogNormBinned.dropna(thresh = 30 )
trainLogNormBinned.shape
#replacing missing categorical values with its own bucket (_NA)
catFeatures = trainLogNormBinned.columns[13:]
for f in catFeatures:
    new_bucket = f+"_NA"
    trainLogNormBinned[f].fillna(new_bucket,inplace = True)
trainLogNormBinned['intFeature2'].mean()
#replacing missing integer values with the mean
intFeatures = trainLogNormBinned.columns[1:13]
for f in intFeatures:
    colMean = trainLogNormBinned[f].mean()
    trainLogNormBinned[f].fillna(colMean,inplace = True)
#write pre-processed data to file for later use
trainLogNormBinned.to_csv('eda/pre-processed.csv', index = False)
#read pre-processed data

#split processed data into train and test set
from sklearn.model_selection import train_test_split
y = trainLogNormBinned.label
X = trainLogNormBinned.drop(columns = ['label'])
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.20, random_state=11)
#write to file train and test data for later use
X_train.to_csv('eda/X_train.csv', index = False)
X_test.to_csv('eda/X_test.csv', index = False)
y_train.to_csv('eda/y_train.csv', index = False)
y_test.to_csv('eda/y_test.csv', index = False)
intFeatures = trainLogNormBinned.columns[1:13]
catFeatures = trainLogNormBinned.columns[13:]
target = trainLogNormBinned.columns[0]<jupyter_output><empty_output><jupyter_text>Taking a Go at Logistic Regression as is on Pre-Processed Data  Notes on the preprocessed data* dropped intFeature12 and catFeature22 based on columnwise incompleteness* dropped rows missing 3/4 or more of total features values ~at least 30 columns needed to have data* converted negative integer values in intFeature2 to absolute value (assumed negative was input error)* log transformed and normalized integer features* bucketed levels/labels that have fewer than 30 observations into a single level/label per category feature (ends with _binned)* imputed NaN values:    *created a new label/level for categorical NaN values    *imputed mean of column for int features    Resources:- [see logistic regression section](https://spark.apache.org/docs/latest/mllib-linear-methods.html)<jupyter_code>intFeatures = ['intFeature1', 'intFeature2', 'intFeature3', 'intFeature4',
       'intFeature5', 'intFeature6', 'intFeature7', 'intFeature8',
       'intFeature9', 'intFeature10', 'intFeature11', 'intFeature13']
catFeatures = ['catFeature3_binned', 'catFeature1_binned', 'catFeature2_binned',
       'catFeature4_binned', 'catFeature5_binned', 'catFeature6_binned',
       'catFeature7_binned', 'catFeature8_binned', 'catFeature9_binned',
       'catFeature10_binned', 'catFeature11_binned', 'catFeature12_binned',
       'catFeature13_binned', 'catFeature14_binned', 'catFeature15_binned',
       'catFeature16_binned', 'catFeature17_binned', 'catFeature18_binned',
       'catFeature19_binned', 'catFeature20_binned', 'catFeature21_binned',
       'catFeature23_binned', 'catFeature24_binned', 'catFeature25_binned',
       'catFeature26_binned']

target = ['click']
#setting schema and reading in pre-processed data to pyspark dataframe
outcomeField = [StructField("click", IntegerType(), True)]
quantFields = [StructField(f, DoubleType(), True) for f in intFeatures]
qualFields = [StructField(f, StringType(), True) for f in catFeatures]
schema = StructType(outcomeField + quantFields + qualFields)

df = spark.read.schema(schema) \
    .option("ignoreLeadingWhiteSpace",True) \
    .option("ignoreTrailingWhiteSpace",True) \
    .csv('eda/pre-processed.csv',nullValue = "NaN")
catCols = [f for f in df.columns if "cat" in f]
numericCols = [f for f in df.columns if "int" in f]
target = ['click']
indexers = [StringIndexer(inputCol=c, outputCol="{0}_indexed".format(c), handleInvalid="skip") for c in catCols]
label = StringIndexer(inputCol= target[0], outputCol="label", handleInvalid="skip")
#formatting for logistic regression
encoder = OneHotEncoderEstimator(
    inputCols=[indexer.getOutputCol() for indexer in indexers],
    outputCols=["{0}_encoded".format(indexer.getOutputCol()) for indexer in indexers],
    dropLast=False)

assembler = VectorAssembler(
    inputCols= encoder.getOutputCols() + numericCols,
    outputCol= "features")

pipeline = Pipeline(stages = indexers + [encoder,assembler,label])

processedDF = pipeline.fit(df).transform(df)
# Keep relevant columns
selectedcols = ["label", "features"]
dataset = processedDF.select(selectedcols)
display(dataset)
### Randomly split data into training and test sets. set seed for reproducibility
(trainingData, testData) = dataset.randomSplit([0.7, 0.3], seed=100)
print(trainingData.count())
print(testData.count())
# Create initial LogisticRegression model
from pyspark.ml.classification import LogisticRegression
lr = LogisticRegression(labelCol="label", featuresCol="features", maxIter=10)
# Train model with Training Data
lrModel = lr.fit(trainingData)
# Make predictions on test data using the transform() method. - LogisticRegression.transform() will only use the 'features' column.
predictions = lrModel.transform(testData)
# View model's predictions and probabilities of each prediction class
# You can select any columns granted they're in the schema
selected = predictions.select("label", "prediction", "probability")
selected.show(5)<jupyter_output>+-----+----------+--------------------+
|label|prediction|         probability|
+-----+----------+--------------------+
|  0.0|       0.0|[0.92870709566892...|
|  0.0|       0.0|[0.85836635788769...|
|  0.0|       0.0|[0.89089600238478...|
|  0.0|       0.0|[0.85420658571523...|
|  0.0|       0.0|[0.61823702911393...|
+-----+----------+--------------------+
only showing top 5 rows<jupyter_text>Evaluation of Logistic Regression Model Fit[scikit LogLoss](https://scikit-learn.org/stable/modules/model_evaluation.htmllog-loss)[when accuracy doesn't make sense](https://towardsdatascience.com/metrics-to-evaluate-your-machine-learning-algorithm-f10ba6e38234)[good site for log loss](https://deeplearningretreat.com/understanding-binary-cross-entropy-log-loss-a-visual-explanation/) Accuracy vs LogLoss(cross-entropy loss)* Accuracy is the number of predictions correct out of all predictions. Accuracy can be a flawed evaluation metric when there is a sizable discrepancy between the number of samples belonging to each class. In the context of our dataset, roughly 3/4 of the data belong to our "no-click" class (0) and 1/4 of the sample belongs to our "click" class (1). Given that ratio, our model can easily get 75% training accuracy by simply predicting every training sample belonging to the "no-click" class. Accuracy tends to work well only when there are a near equal number of samples belonging to each class. Further, accuracy gives no context with respect to costs. It may be much more costly to make false positives than false negatives. Accuracy doesn't give us a means to make that distinction.* Log loss takes into account the uncertainty of predictions based on how much they vary from the actual label. Log loss penalizes both types of errors, but especially those predications that are confident (higher or lower probabilities) and wrong. Our logistic regression model outputs probabilities for each label (no click, click) so we can use those to compute the log loss of our model. A log loss close to zero implies that are probabilities are both very confident and accurate. So, the lower the log loss, the better the fit. for a single sample/record, log loss is defined as the negative log-likelihood of the classifier given the true label:$L_{\log}(y, p) = -\log \operatorname{Pr}(y|p) = -(y \log (p) + (1 - y) \log (1 - p))$where $p = \operatorname{Pr}(y = 1)$ and $y \in \{0,1\}$ is the true labeland for the entire dataset:$L_{\log}(Y, P) = -\log \operatorname{Pr}(Y|P) = - \frac{1}{N} \sum_{i=0}^{N-1} \sum_{k=0}^{K-1} y_{i,k} \log p_{i,k}$<jupyter_code>#converting to pandas dataframe...has to be a better way to do this with just pyspark
mod_output = selected.toPandas()
mod_output.head(5)
#pyspark doesn't have a logLoss evaluation method in its Evaluator class so we need to write our own function

import math

def logLoss(df):
    '''takes the pandas dataframe resulting from fitting our test set
    against our logistic regression model'''
    logLoss_vals = []
    epsilon = 1e-16
    for idx in df.index:
        if df.label[idx] == 1:
            #the 2nd element in the list (index 1) is the probability of click
            loss = -math.log(df.probability[idx][1] + epsilon)
        else:
            loss = -math.log(1. - df.probability[idx][1] + epsilon)
        
        logLoss_vals.append(loss)
        
    return np.mean(logLoss_vals)

logLoss(mod_output)<jupyter_output><empty_output><jupyter_text>initial model resulted in a LogLoss of .47<jupyter_code>from pyspark.sql.functions import udf
from pyspark.sql.types import FloatType
#need a user defined function to manipulate columns of array types
firstelement=udf(lambda v:float(v[1]),FloatType())
selected.select(firstelement('probability')).show(5)
#pyspark implementation for logLoss
pyspark_mod = selected.withColumn(
        'logloss'
        , -f.col('label')*f.log(firstelement(f.col('probability'))) - (1.-f.col('label'))*f.log(1.-firstelement(f.col('probability')))
    )
pyspark_modLL = pyspark_mod.agg(f.mean('logloss').alias('ll')).collect()[0]['ll']
pyspark_modLL
from pyspark.ml.evaluation import BinaryClassificationEvaluator
# Evaluate model accuracy (default is area under ROC curve)
evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction")
evaluator.evaluate(predictions)<jupyter_output><empty_output><jupyter_text>manual feature selection (see preprocessed notes) with binary logistic regression yielded accuracy of .775  Regularizing with LassoLasso shrinks the less important feature’s coefficient to zero, and therefore, may remove some feature altogether. So, this can be another form of feature selection when we have a large number of features[see reference](https://spark.apache.org/docs/1.5.2/ml-linear-methods.html)<jupyter_code>#logistic regression with lasso regularization
lr_l1 = LogisticRegression(labelCol="label", featuresCol="features", regParam=0.0, elasticNetParam =1.0, maxIter=10)
# Fit the model
lrModel_lasso = lr_l1.fit(trainingData)
predictions_lasso = lrModel_lasso.transform(testData)
evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction")
evaluator.evaluate(predictions_lasso)<jupyter_output><empty_output><jupyter_text>Testing Fit after Dropping intFeature11removing near perfect collinearity in the dataset<jupyter_code>#reading in the pre-processed data
#setting schema and reading in pre-processed data to pyspark dataframe
outcomeField = [StructField("click", IntegerType(), True)]
quantFields = [StructField(f, DoubleType(), True) for f in intFeatures]
qualFields = [StructField(f, StringType(), True) for f in catFeatures]
schema = StructType(outcomeField + quantFields + qualFields)

df = spark.read.schema(schema) \
    .option("ignoreLeadingWhiteSpace",True) \
    .option("ignoreTrailingWhiteSpace",True) \
    .csv('eda/pre-processed.csv',nullValue = "NaN")
#drop feature that is highly correlated with intFeature7
df = df.drop('intFeature11')
#prep for training
catCols = [f for f in df.columns if "cat" in f]
numericCols = [f for f in df.columns if "int" in f]
target = ['click']
indexers = [StringIndexer(inputCol=c, outputCol="{0}_indexed".format(c), handleInvalid="skip") for c in catCols]
label = StringIndexer(inputCol= target[0], outputCol="label", handleInvalid="skip")
#formatting for logistic regression
encoder = OneHotEncoderEstimator(
    inputCols=[indexer.getOutputCol() for indexer in indexers],
    outputCols=["{0}_encoded".format(indexer.getOutputCol()) for indexer in indexers],
    dropLast=False)

assembler = VectorAssembler(
    inputCols= encoder.getOutputCols() + numericCols,
    outputCol= "features")

pipeline = Pipeline(stages = indexers + [encoder,assembler,label])

processedDF = pipeline.fit(df).transform(df)
# Keep relevant columns and split for train/test
selectedcols = ["label", "features"]
dataset = processedDF.select(selectedcols)
### Randomly split data into training and test sets. set seed for reproducibility
(trainingData, testData) = dataset.randomSplit([0.7, 0.3], seed=100)
# Create initial LogisticRegression model
from pyspark.ml.classification import LogisticRegression
lr = LogisticRegression(labelCol="label", featuresCol="features", maxIter=10)
# Train model with Training Data
lrModel = lr.fit(trainingData)
# Make predictions on test data using the transform() method. - LogisticRegression.transform() will only use the 'features' column.
predictions = lrModel.transform(testData)
# You can select any columns granted they're in the schema
selected = predictions.select("label", "prediction", "probability")
#evaluation of logit fit after dropping intFeature 11 with logLoss
from pyspark.sql.functions import udf
from pyspark.sql.types import FloatType
from pyspark.sql import functions as f
#need a user defined function to manipulate columns of array types
firstelement=udf(lambda v:float(v[1]),FloatType())

#pyspark implementation for logLoss
pyspark_mod = selected.withColumn(
        'logloss'
        , -f.col('label')*f.log(firstelement(f.col('probability'))) - (1.-f.col('label'))*f.log(1.-firstelement(f.col('probability')))
    )


pyspark_mod.agg(f.mean('logloss').alias('ll')).collect()[0]['ll']<jupyter_output><empty_output><jupyter_text>Binning the Int Features data as a method of feature engineeringWe saw that the integer features were still very skewed after log-transforming and normalizing them. Our model may benefit from improving the distribution of those<jupyter_code>#load the trainSample into memory with a pandas dataframe
import glob
files = np.sort(glob.glob('eda/trainSample/*.csv'))
trainSample = pd.concat([pd.read_csv(f) for f in files],ignore_index = True)
#replace negative values with NaNs in intFeature2
f2 = trainSample['intFeature2']
neg_vals = f2[f2<0].unique()
trainSample.intFeature2 = trainSample['intFeature2'].replace(to_replace = neg_vals, value = np.NaN)
#drop columns missing most of their data; drop rows missing most of their data
# trainSample = trainSample.drop(columns = ['intFeature12','catFeature22']) \
#                 .dropna(thresh = 30 )
#impute mean on int Features
intFeatures = trainSample.columns[1:13]
for f in intFeatures:
    colMean = trainSample[f].mean()
    trainSample[f].fillna(colMean,inplace = True)
    
trainSample.head(5)
#iteratively build dataframe with quantile binning on intFeatures

for f in intFeatures:
    newCol_name = f + "_Qbin"
    trainSample[newCol_name] = pd.qcut(trainSample[f].rank(method='first'), 4)

trainSample.head(5)
#iteratively build dataframe with categorical adjustments
catFeatures = trainSample.columns[13:38]
for f in catFeatures:
    newCol_name = f + "_binned"
    row_elements = []
    #counts of unique labels per category
    uniq = trainSample[f].value_counts()
    #indx of labels with fewer than 30 observations (return the labels with .index)
    idx = set(uniq[uniq<30].index)
    labels = trainSample[f].values
    for label in labels:
        if label in idx:
            row_elements.append(f+"-bucket")
        else:
            row_elements.append(label)
            
    trainSample[newCol_name] = row_elements
    
trainSample.head(5)
all_catFeatures = trainSample.columns.str.contains('cat')
all_catFeatures = trainSample.columns[catFeatures]
all_catFeatures
#add NA buckets...can do this in the above code when building columns iteratively
for f in all_catFeatures:
    new_bucket = f+"_NA"
    trainSample[f].fillna(new_bucket,inplace = True)
#created dataframe of all binned features
binned = trainSample.columns.str.contains('bin|label')

trainSample_binned = trainSample.columns[binned]
trainSample_bins = trainSample[trainSample_binned]
trainSample_bins.head(5)
#write to file for later use
fpath = 'eda/trainSampleBinned.csv'
trainSample_bins.to_csv(path_or_buf = fpath, index = False)
#setting up schema to read in pyspark dataframe
catFeatures_bin = ['intFeature1_Qbin', 'intFeature2_Qbin', 'intFeature3_Qbin',
       'intFeature4_Qbin', 'intFeature5_Qbin', 'intFeature6_Qbin',
       'intFeature7_Qbin', 'intFeature8_Qbin', 'intFeature9_Qbin',
       'intFeature10_Qbin', 'intFeature11_Qbin', 'intFeature13_Qbin',
       'catFeature1_binned', 'catFeature2_binned', 'catFeature3_binned',
       'catFeature4_binned', 'catFeature5_binned', 'catFeature6_binned',
       'catFeature7_binned', 'catFeature8_binned', 'catFeature9_binned',
       'catFeature10_binned', 'catFeature11_binned', 'catFeature12_binned',
       'catFeature13_binned', 'catFeature14_binned', 'catFeature15_binned',
       'catFeature16_binned', 'catFeature17_binned', 'catFeature18_binned',
       'catFeature19_binned', 'catFeature20_binned', 'catFeature21_binned',
       'catFeature23_binned', 'catFeature24_binned', 'catFeature25_binned',
       'catFeature26_binned']

target = ['click']

#setting schema and reading in pre-processed data to pyspark dataframe
outcomeField = [StructField("click", IntegerType(), True)]
qualFields = [StructField(f, StringType(), True) for f in catFeatures_bin]
schema = StructType(outcomeField + qualFields)

df = spark.read.schema(schema) \
    .option("ignoreLeadingWhiteSpace",True) \
    .option("ignoreTrailingWhiteSpace",True) \
    .csv('eda/trainSampleBinned.csv',nullValue = "NaN")
df.show(5)
df.columns[1:]
catCols = df.columns[1:]
target = ['click']
indexers = [StringIndexer(inputCol=c, outputCol="{0}_indexed".format(c), handleInvalid="skip") for c in catCols]
label = StringIndexer(inputCol= target[0], outputCol="label", handleInvalid="skip")
#formatting for logistic regression
encoder = OneHotEncoderEstimator(
    inputCols=[indexer.getOutputCol() for indexer in indexers],
    outputCols=["{0}_encoded".format(indexer.getOutputCol()) for indexer in indexers],
    dropLast=False)

assembler = VectorAssembler(
    inputCols= encoder.getOutputCols(),
    outputCol= "features")

pipeline = Pipeline(stages = indexers + [encoder,assembler,label])

processedDF = pipeline.fit(df).transform(df)
processedDF.show(5)
# Keep relevant columns and split for train/test
selectedcols = ["label", "features"]
dataset = processedDF.select(selectedcols)
### Randomly split data into training and test sets. set seed for reproducibility
(trainingData, testData) = dataset.randomSplit([0.7, 0.3], seed=100)
# Create initial LogisticRegression model
from pyspark.ml.classification import LogisticRegression
lr = LogisticRegression(labelCol="label", featuresCol="features", maxIter=10)
# Train model with Training Data
lrModel = lr.fit(trainingData)
# Make predictions on test data using the transform() method. - LogisticRegression.transform() will only use the 'features' column.
predictions = lrModel.transform(testData)
# You can select any columns granted they're in the schema
selected = predictions.select("label", "prediction", "probability")
#evaluation of logit fit after dropping intFeature 11 with logLoss
from pyspark.sql.functions import udf
from pyspark.sql.types import FloatType
from pyspark.sql import functions as f
#need a user defined function to manipulate columns of array types
firstelement=udf(lambda v:float(v[1]),FloatType())

#pyspark implementation for logLoss
pyspark_mod = selected.withColumn(
        'logloss'
        , -f.col('label')*f.log(firstelement(f.col('probability'))) - (1.-f.col('label'))*f.log(1.-firstelement(f.col('probability')))
    )


pyspark_mod.agg(f.mean('logloss').alias('ll')).collect()[0]['ll']
##---------------__###
# from pyspark.ml.feature import Bucketizer

# splits = [-float("inf"), -0.5, 0.0, 0.5, float("inf")]

# data = [(-999.9,), (-0.5,), (-0.3,), (0.0,), (0.2,), (999.9,)]
# dataFrame = spark.createDataFrame(data, ["features"])

# bucketizer = Bucketizer(splits=splits, inputCol="features", outputCol="bucketedFeatures")

# # Transform original data into its bucket index.
# bucketedData = bucketizer.transform(dataFrame)

# print("Bucketizer output with %d buckets" % (len(bucketizer.getSplits())-1))
# bucketedData.show()

#---for pyspark---#
# from pyspark.ml.feature import Bucketizer
# bucketizer = Bucketizer(splits=[ 0, 6, 18, 60, float('Inf') ],inputCol="ages", outputCol="buckets")
# df_buck = bucketizer.setHandleInvalid("keep").transform(df)

# df_buck.show()
def log_transform(x):
    '''shifts x by un'''
    return np.log(x+1)
x = pd.value_counts(pd.cut(f9,15))

# ax = x.plot.bar(rot=0, color="b", figsize=(6,4))
# ax.set_xticklabels([c[1:-1].replace(","," to") for c in x.cat.categories])
# plt.show()
f9.apply(lambda x: np.log(x+1)).hist()<jupyter_output><empty_output><jupyter_text>--------------------------------------<jupyter_code># load the raw training data into an RDD then split into a training and validation dataset
trainDataRDD = sc.textFile('data/train.txt')
trainRDD, validationRDD = trainDataRDD.randomSplit([0.8,0.2], seed = 1)
#will need to sample the data for EDA since we have such a large dataset (starting with 450K observations ~1% of all observations)
trainSample = trainRDD.map(lambda x: x.split('\t')) \
                .takeSample(False, 100000)
#transform ndarray to dataframe, add headers, and save to csv for later use
cols = ['outcome']+["var"+str(i+1) for i in np.arange(39)]
df = pd.DataFrame(trainSample,columns = cols)
df.to_csv(PWD + '/data/trainSample.csv', index = False)
#read in the csv data - this automatically converts datatypes to floats and ints
df = pd.read_csv('data/trainSample.csv')
#summary of our quantitative variables
#summary = df.describe()<jupyter_output><empty_output><jupyter_text>>_ An initial look at the spread of each our our quantitative variables from our sample suggests we'll need to normalize these variables. We can also see that several quantitative variables have relatively few observations so sparsity will be a concern. >_ we can see from our categorical variables that there are many levels within each category. For instance var16, var25, var34 all have over 40,000 levels. If we were to make indicator variables out of these, in the context of a logistic/binary regression, then we'll have a very high dimensional dataset.  Initial Thoughts on Concepts* a lot of work during the preprocess phase* normalization* sparsity* regularization for feature selection in high dimensionality * binary logistic regression, its assumptions and the bias variance tradeoff* scalability<jupyter_code>#helper functions
#-columns are tab-separated

#note: to convert RDD to DataFrame in Pyspark, each record (line) needs to be converted to a row object
#a quick way to do this is to convert the record to a tuple

def Row_formatter(line):
    record = line.split('\t')
    return tuple(record)

df = trainDataRDD.map(Row_formatter).toDF()
df.show(1)
col_names = df.columns

df.select('_15').show(5)
df.count()
#number of records in the train data
trainDataRDD.map(formatter).count()
records = trainDataRDD.map(formatter)

trainDF = spark.createDataFrame(trainData)<jupyter_output><empty_output><jupyter_text>total of 45,840,617  observations in the train data  PySpark Code to processing and model data<jupyter_code># 13 features taking integer values and 26 taking categorical values (we can't assume ordinal) - making total of 39 features + 1 outcome
outcomeField = [StructField("label", IntegerType(), True)]
quantFields = [StructField("intFeature"+str(i+1), IntegerType(), True) for i in np.arange(13)]
qualFields = [StructField("catFeature"+str(i+1), StringType(), True) for i in np.arange(26)]
schema = StructType(outcomeField + quantFields + qualFields)

df = spark.read \
    .schema(schema) \
    .format("csv") \
    .option("header", "true") \
    .load("eda/trainSample/*.csv")
#clean up: drop duplicate rows; rows missing at least 30 out of 39 values (>= 3/4 of row-wise data); columns missing >= 3/4 of data 
drop_cols = ['intFeature12','catFeature22']
df = df.dropDuplicates() \
    .dropna(thresh=30) \
    .drop(*drop_cols)
#-----------imputing missing values---------------#
#bucket missing values in categorical data into its own sub-category for all releveant columns
categoryCols = [col for col in  df.columns if 'cat' in col]
df = df.fillna("NA_Bucket", subset=categoryCols)
#assuming negative values in intFeatures were incorrectly coded and should be null
neg_overide = f.udf(lambda val: None if val < 0 else val, IntegerType())
df = df.withColumn('intFeature2', neg_overide(df.intFeature2))<jupyter_output><empty_output><jupyter_text>--- Imputing Missing Values with Pyspark -----<jupyter_code>#update categorical column values conditioned on the number of observations for each value

for c in categoryCols:
    
    #collect unique values with greater than 30 observations
    bucket = set(df.groupBy(c) \
    .count() \
    .filter(f.col('count') > 30) \
    .select(f.col(c)) \
    .rdd \
    .flatMap(lambda x: x) \
    .collect())
    
    #update dataframe with rare values (fewer than 30) as their own bucketed value
    df = df.withColumn(c, 
                   f.when(f.col(c).isin(bucket)
                        , f.col(c)).otherwise("Rare_Bucket"))
#update integer columns missing values with the column mean

integerCols = [col for col in  df.columns if 'int' in col]
mean_dict = {}
for c in integerCols:
    #compute mean for each column
    col_mean = df.select(f.mean(f.col(c)).alias('mean')).collect()[0]['mean']
    mean_dict[c] = col_mean

#update null values in int columns:
df = df.na.fill(mean_dict)
#log transform int variables
for c in integerCols:
    df = df.withColumn(c, 
                   f.log(f.col(c)+1))
# normalize
for c in integerCols:
    #compute mean for each column
    col_mean = df.select(f.mean(f.col(c)).alias('mean')).collect()[0]['mean']
    col_std = df.select(f.stddev(f.col(c)).alias('std')).collect()[0]['std']
    #update dataframe
    df = df.withColumn(c,(f.col(c) - col_mean)/col_std)
# write df to file for later use
df.write \
    .option("header", "true") \
    .csv("data/processed")
#setting schema and reading in pre-processed data to pyspark dataframe

integerCols = ['intFeature1', 'intFeature2', 'intFeature3', 'intFeature4',
       'intFeature5', 'intFeature6', 'intFeature7', 'intFeature8',
       'intFeature9', 'intFeature10', 'intFeature11', 'intFeature13']
       
categoryCols = ['catFeature1', 'catFeature2','catFeature3',
       'catFeature4', 'catFeature5', 'catFeature6',
       'catFeature7', 'catFeature8', 'catFeature9',
       'catFeature10', 'catFeature11', 'catFeature12',
       'catFeature13', 'catFeature14', 'catFeature15',
       'catFeature16', 'catFeature17', 'catFeature18',
       'catFeature19', 'catFeature20', 'catFeature21',
       'catFeature23', 'catFeature24', 'catFeature25',
       'catFeature26']

outcomeField = [StructField("click", IntegerType(), True)]
quantFields = [StructField(f, DoubleType(), True) for f in integerCols]
qualFields = [StructField(f, StringType(), True) for f in categoryCols]
schema = StructType(outcomeField + quantFields + qualFields)


df = spark.read \
    .schema(schema) \
    .format("csv") \
    .option("header", "true") \
    .load("data/processed/*.csv")
target = ['click']
indexers = [StringIndexer(inputCol=c, outputCol="{0}_indexed".format(c), handleInvalid="skip") for c in categoryCols]
label = StringIndexer(inputCol= target[0], outputCol="label", handleInvalid="skip")
#formatting for logistic regression
encoder = OneHotEncoderEstimator(
    inputCols=[indexer.getOutputCol() for indexer in indexers],
    outputCols=["{0}_encoded".format(indexer.getOutputCol()) for indexer in indexers],
    dropLast=False)

assembler = VectorAssembler(
    inputCols= encoder.getOutputCols() + integerCols,
    outputCol= "features")

pipeline = Pipeline(stages = indexers + [encoder,assembler,label])

processedDF = pipeline.fit(df).transform(df)
# Keep relevant columns and split for train/test
selectedcols = ["label", "features"]
dataset = processedDF.select(selectedcols)
### Randomly split data into training and test sets. set seed for reproducibility
(trainingData, testData) = dataset.randomSplit([0.7, 0.3], seed=100)
# Create initial LogisticRegression model
from pyspark.ml.classification import LogisticRegression
lr = LogisticRegression(labelCol="label", featuresCol="features", maxIter=10)
# Train model with Training Data
lrModel = lr.fit(trainingData)
# Make predictions on test data using the transform() method. - LogisticRegression.transform() will only use the 'features' column.
predictions = lrModel.transform(testData)
# You can select any columns granted they're in the schema
selected = predictions.select("label", "prediction", "probability")
#evaluation of logit fit after dropping intFeature 11 with logLoss
from pyspark.sql.functions import udf
from pyspark.sql.types import FloatType
from pyspark.sql import functions as f
#need a user defined function to manipulate columns of array types
firstelement=udf(lambda v:float(v[1]),FloatType())

#pyspark implementation for logLoss
pyspark_mod = selected.withColumn(
        'logloss'
        , -f.col('label')*f.log(firstelement(f.col('probability'))) - (1.-f.col('label'))*f.log(1.-firstelement(f.col('probability')))
    )


pyspark_mod.agg(f.mean('logloss').alias('ll')).collect()[0]['ll']<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11202632416940876, 'fraction_numerical': 0.01957329370555143, 'mean_word_length': 5.099443221546031, 'pattern_counts': {'":': 0, '<': 96, '<?xml version=': 0, '>': 92, 'https://': 8, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '22', 'dir': 'jupyter-structured-clean-dedup', 'id': '155693', 'license': 'MIT', 'n_tokens_mistral': 14827, 'n_tokens_neox': 14046, 'n_words': 4433, 'path': 'criteo_project.ipynb', 'repo_name': 'PeteTrenk/W261-SP19-Team13-FinalProject', 'size': '47105'}