<jupyter_start><jupyter_text>ERDhttps://drive.google.com/open?id=1dHAdBT84rEDf3WiE7FSfFrpnGQVL0wUSsWAGEH7ywZg<jupyter_code>from datetime import datetime
import matplotlib.pyplot as plt

import pyspark.sql.functions as F
from pyspark.sql.window import Window
import numpy as np
import pandas as pd
from sklearn import metrics


pd.options.display.max_columns = 1000
pd.options.display.max_rows = 10

fast_mode = True

%matplotlib inline


from IPython.core.magic import register_line_magic


@register_line_magic
def show(line, n = 5):
    return eval(line).limit(n).toPandas()

@register_line_magic
def sql(line, n = 10):
    return spark.sql(line)
base_path = "/data/kaggle/outbrain_ctr/parquet/"

def cache_df(df, name, sorage_level = StorageLevel.MEMORY_ONLY):
    df.createOrReplaceTempView(name)
    spark.catalog.cacheTable(name)

def load(name, rebase_timestamp = False, cache = True):
    df = spark.read.load(base_path + name)
        
    if rebase_timestamp and "timestamp" in df.columns:
        df = df.withColumn("timestamp"
            , F.expr("cast(from_unixtime(cast((timestamp + 1465876799998)/1000 as int)) as timestamp)"))
    if cache:
        cache_df(df, name)

    df.alias(name)
    print("Number of partitions for df %s: %d" % (name, df.rdd.getNumPartitions()))
        
    return df
!ls -1 /data/kaggle/outbrain_ctr/parquet/<jupyter_output>clicks_test
clicks_train
df_null_removed
df_null_removed2
documents_categories
documents_entities
documents_meta
documents_topics
events
lrModel
lrModel2
merged_enriched
page_views
page_views_sample
promoted_content
sample_submission<jupyter_text>Clicks<jupyter_code>clicks_train = load("clicks_train")
clicks_train.show()
clicks_test = load("clicks_test")
clicks_test.show()
clicks_train.count(), clicks_test.count()<jupyter_output><empty_output><jupyter_text>Distinct count of ad_id in training and test dataset<jupyter_code>%time clicks_train.select("ad_id").distinct().count(), clicks_test.select("ad_id").distinct().count()<jupyter_output>CPU times: user 0 ns, sys: 4 ms, total: 4 ms
Wall time: 2.61 s<jupyter_text>Common ad_id in training and test datasets<jupyter_code>%time clicks_train.select("ad_id").intersect(clicks_test.select("ad_id")).count()
1- 316035/381385<jupyter_output><empty_output><jupyter_text>17% of ad_id in testing dataset are unique. Calculate CTR on training dataset. Note, we cannot calculate the CTR on test dataset since clicked column is provded the value. In fact, the rask is to predict the probability of click.<jupyter_code>ctrs = clicks_train.groupBy("ad_id")\
.agg(F.expr("sum(clicked)/count(*)").alias("ctr"), F.count("*").alias("view_count"))

ctrs.show()
%time ctrs.select("ctr").describe().show()<jupyter_output>+-------+-------------------+
|summary|                ctr|
+-------+-------------------+
|  count|             478950|
|   mean| 0.1432241011540594|
| stddev|0.21768444463682177|
|    min|                0.0|
|    max|                1.0|
+-------+-------------------+

CPU times: user 4 ms, sys: 0 ns, total: 4 ms
Wall time: 1.81 s<jupyter_text>Median CTR<jupyter_code>%time ctrs.selectExpr("percentile(ctr, 0.5)").show()
ctrs.filter("ad_id = 182320").show()<jupyter_output>+------+------------------+----------+
| ad_id|               ctr|view_count|
+------+------------------+----------+
|182320|0.7308917197452229|       628|
+------+------------------+----------+<jupyter_text>Find 99, 95 and 90 percentile values of the view counts of the ads.<jupyter_code>view_counts = ctrs.select("view_count").toPandas()
np.percentile(view_counts["view_count"], [99, 95, 90])<jupyter_output><empty_output><jupyter_text>To build confidence on the CTR, filter out the ads with fewer than 100 views (approx 99 percentile value)<jupyter_code>%time ctrs.filter("view_count>100").select("ctr").toPandas()["ctr"].plot.hist(bins = 50, density = True)
plt.xlabel("CTR")
plt.ylabel("Frquency (normalized)")<jupyter_output>CPU times: user 628 ms, sys: 24 ms, total: 652 ms
Wall time: 1.9 s<jupyter_text>Consier, the CTR as baseline for click prediction. Using CTR as based, calculate the MAP (mean avg precision)<jupyter_code>#y_pred = clicks_train.join(ctrs.select("ad_id", "ctr"), on = ["ad_id"], how="left").select("ctr").toPandas()["ctr"]
#y_true = clicks_train.select("clicked").toPandas()["clicked"]

#%time metrics.average_precision_score(y_true, y_pred)
clicks_test_baseline = clicks_test.join(ctrs.select("ad_id", "ctr"), on = ["ad_id"], how="left")
clicks_test_baseline.show()
clicks_test_baseline.groupBy("display_id").count()
clicks_test_baseline.groupBy("display_id").agg(F.sum("ctr").alias("ap")).selectExpr("avg(ap)").show()<jupyter_output>+------------------+
|           avg(ap)|
+------------------+
|0.9835441908732905|
+------------------+<jupyter_text>How many ads are there for each display_id?<jupyter_code>%time clicks_train.groupBy("display_id").count().select("count").distinct().show()<jupyter_output>+-----+
|count|
+-----+
|    9|
|    3|
|    6|
|    7|
|    5|
|    8|
|   11|
|    4|
|    2|
|   12|
|   10|
+-----+

CPU times: user 4 ms, sys: 0 ns, total: 4 ms
Wall time: 4.03 s<jupyter_text>Does each display_id in the training dataset has atleast one click?<jupyter_code>%time clicks_train.groupBy("display_id").agg(F.sum("clicked").alias("clicks")).filter("clicks=0").count()<jupyter_output>CPU times: user 4 ms, sys: 0 ns, total: 4 ms
Wall time: 4.31 s<jupyter_text>So, each display_id has atleast one click. Does it display_id in the training dataset has more than click?<jupyter_code>%time clicks_train.groupBy("display_id").agg(F.sum("clicked").alias("clicks")).filter("clicks>1").count()<jupyter_output>CPU times: user 0 ns, sys: 0 ns, total: 0 ns
Wall time: 3.85 s<jupyter_text>So, each display_id in the clicks dataset has only one click.  Page Views<jupyter_code>if fast_mode:
    print("Loading page_view sample dataset")
    page_views = load("page_views_sample", rebase_timestamp=True, cache=True)
    #page_views = page_views.sample(False, 0.01, 1)
    #cache_df(page_views, "page_views")
else:
    print("Loading full page_view dataset")
    page_views = load("page_views", rebase_timestamp=True, cache=False)
page_views.printSchema()
page_views.show()<jupyter_output>+--------------+-----------+-------------------+--------+------------+--------------+
|          uuid|document_id|          timestamp|platform|geo_location|traffic_source|
+--------------+-----------+-------------------+--------+------------+--------------+
|289a8ac9e3b3be|     647313|2016-06-14 23:30:43|       1|       PK>08|             1|
|2cd20abba0d948|     647313|2016-06-14 18:19:16|       3|   US>CA>804|             1|
|6149469c04c102|     647313|2016-06-14 10:12:32|       1|       AU>01|             1|
|f3bca9d885121d|     647313|2016-06-15 06:07:45|       2|   US>KS>616|             1|
|594de491d7a3d1|     647313|2016-06-14 13:09:20|       1|   US>CA>807|             1|
|b5fe86af022f0c|     647313|2016-06-14 12:13:02|       3|       GB>H9|             1|
|56bb88f5111d21|     647313|2016-06-14 11:02:45|       1|       SI>61|             1|
|82a3dae458fd43|     647313|2016-06-14 15:46:58|       1|       MU>17|             1|
|4323bb94758e52|     647313|2016-06-14 10:04:50|      [...]<jupyter_text>Page views table is nearly 100 GB in decompressed csv file. How many records are there?<jupyter_code>page_views.count()<jupyter_output><empty_output><jupyter_text>Does each record in page_views have timestamp?<jupyter_code>%time page_views.filter("isnull(timestamp)").count()
stats = page_views.selectExpr("count(distinct(uuid)) as users"
                      , "count(distinct(document_id)) as documents"
                      , "count(distinct(geo_location)) as locations")
%time stats.show()<jupyter_output>+-------+---------+---------+
|  users|documents|locations|
+-------+---------+---------+
|9202149|    59849|     2964|
+-------+---------+---------+

CPU times: user 4 ms, sys: 0 ns, total: 4 ms
Wall time: 3.37 s<jupyter_text>Some users are more frequent visitor that the others. Find the number of visitors based on the view counts.<jupyter_code>page_views_by_user = page_views.groupBy("uuid").count().groupBy("count").count()\
                    .toDF("view_count", "num_users").toPandas().sort_values("view_count")
page_views_by_user
page_views_by_user.iloc[:20, :].plot.bar("view_count", "num_users")<jupyter_output><empty_output><jupyter_text>Number of unique users<jupyter_code>users_distinct_count = page_views_by_user.num_users.sum()<jupyter_output><empty_output><jupyter_text>Average page views per user<jupyter_code>(page_views_by_user.num_users * page_views_by_user.view_count).sum()/users_distinct_count<jupyter_output><empty_output><jupyter_text>Cumulative Percentual<jupyter_code>page_views_by_user["cum_percentual"] = page_views_by_user.num_users.cumsum()/users_distinct_count
page_views_by_user
page_views_by_user.iloc[:20, :].plot.line(x = "view_count", y = "cum_percentual")
page_views_by_user.tail(10)<jupyter_output><empty_output><jupyter_text>Page views by platform<jupyter_code>page_views_by_platform = page_views.groupBy("platform")\
.count().toPandas().set_index("platform").sort_index()
page_views_by_platform
page_views_by_platform.plot.pie(y = "count", labels = ["Desktop", "Mobile", "Tablet"]
                    , figsize = (8, 8), autopct = "%.2f", fontsize = 15)
plt.title("Page views by platform")<jupyter_output><empty_output><jupyter_text>Page views by traffic source<jupyter_code>page_views_by_traffic_source = page_views.groupBy("traffic_source").count().toPandas()
page_views_by_traffic_source = page_views_by_traffic_source.set_index("traffic_source").sort_index()
page_views_by_traffic_source
page_views_by_traffic_source.plot.pie(y = "count", labels = ["Internal", "Search", "Social"]
                    , figsize = (8, 8), autopct = "%.2f", fontsize = 15)
plt.title("Page views by traffic source")<jupyter_output><empty_output><jupyter_text>Events<jupyter_code>events = load("events")
events.show()
events = events.withColumn("timestamp", F.expr("from_unixtime(cast((timestamp + 1465876799998)/1000 as int))"))
events.show()
events = load("events", rebase_timestamp=True)
events.count()<jupyter_output><empty_output><jupyter_text>Find distinct count of users, document, and location<jupyter_code>%time events.selectExpr("count(distinct uuid)", "count(distinct document_id)", "count(distinct geo_location)").first()<jupyter_output>CPU times: user 4 ms, sys: 0 ns, total: 4 ms
Wall time: 8 s<jupyter_text>Is the display_id unique in the events dataset?<jupyter_code>%time events.groupBy("display_id").count().filter("count>1").count()<jupyter_output>CPU times: user 0 ns, sys: 4 ms, total: 4 ms
Wall time: 4.64 s<jupyter_text>Do all the display_id in the clicks_train present in events?<jupyter_code>%time clicks_train.select("display_id").distinct().join(events, on = ["display_id"], how = "left_anti").count()<jupyter_output>CPU times: user 0 ns, sys: 4 ms, total: 4 ms
Wall time: 5.68 s<jupyter_text>So, display_id for each record in clicks_train is present in events dataset.Check the same for clicks_test dataset.<jupyter_code>%time clicks_test.select("display_id").distinct().join(events, on = ["display_id"], how = "left_anti").count()<jupyter_output>CPU times: user 0 ns, sys: 0 ns, total: 0 ns
Wall time: 3.21 s<jupyter_text>Average events by user<jupyter_code>events.selectExpr("count(*)/count(distinct uuid) avg_event_by_user").first()<jupyter_output><empty_output><jupyter_text>Does the timestamp exist for each record in events?<jupyter_code>events.filter("isnull(timestamp)").count()<jupyter_output><empty_output><jupyter_text>Does the timestamp in clicks_train and clicks_test have overlap?<jupyter_code>%time clicks_train.join(events, on = ["display_id"]).select("timestamp").describe().show()
%time clicks_test.join(events, on = ["display_id"]).select("timestamp").describe().show()<jupyter_output>+-------+
|summary|
+-------+
|  count|
|   mean|
| stddev|
|    min|
|    max|
+-------+

CPU times: user 4 ms, sys: 0 ns, total: 4 ms
Wall time: 31.5 ms<jupyter_text>Clearly, the date ranges of training and test data overlap.  Alignment between Events and Page Views Number of page views without matching event How many events have matching page views by uuid and document_id How many events have no page views How many views records have not matching events  A given user might visit the same page more than once. Show sample events for which multiple page_views exist.   Show the distribution of number of ditinct users who view the same document multiple times<jupyter_code>def join_views_and_events(columns):
    df1 = page_views.select(*columns).withColumn("page_views", F.lit(1)).withColumn("events", F.lit(0))
    df2 = events.select(*columns).withColumn("page_views", F.lit(0)).withColumn("events", F.lit(1))
    df3 = df1.union(df2)
    df4 = df3.groupBy(columns).agg(
                F.sum("page_views").alias("page_views_count"), 
                F.sum("events").alias("events_count"))
    return df4
# Cache output to disk. The dataframe is too large to hold in the memory of the current machine 
views_and_event = join_views_and_events(["uuid", "document_id"]).persist(StorageLevel.DISK_ONLY)
views_and_event.show()<jupyter_output>+--------------+-----------+----------------+------------+
|          uuid|document_id|page_views_count|events_count|
+--------------+-----------+----------------+------------+
|3286c54a862512|     647776|               1|           0|
|a29f99948e908e|     647776|               1|           0|
|1dd38cfc32d0af|     647776|               1|           0|
|2ebe105af88084|     647776|               1|           0|
|fc4f1e271ab611|     647776|               1|           0|
|7a14dda54e7f60|     649004|               1|           0|
|7c04b579125457|     649161|               1|           0|
|df1f16aaeff918|     649161|               1|           0|
|7932260aef663f|     649395|               1|           0|
|b78c5990704b55|     649395|               1|           0|
|c9cc4498d587e5|     649395|               1|           0|
|d0f57454f9bd8d|     649395|               1|           0|
|514bdfac720462|     649395|               1|           0|
|5e981a9aa49864|     649707|               1|           [...]<jupyter_text>How many user-document combination does not have any click on ads?<jupyter_code>%time views_and_event.filter("page_views_count > 0 and events_count = 0").count()<jupyter_output>CPU times: user 4 ms, sys: 0 ns, total: 4 ms
Wall time: 10.2 s<jupyter_text>How many user-doc combination does have machine record in page views?<jupyter_code>%time views_and_event.filter("page_views_count = 0 and events_count > 0").count()<jupyter_output>CPU times: user 0 ns, sys: 0 ns, total: 0 ns
Wall time: 367 ms<jupyter_text>Considering events represents the page views that have got clicks, what fraction of page views got clicks?<jupyter_code>events.count()/page_views.count()<jupyter_output><empty_output><jupyter_text>Let's take a sample uuid and document_id to see whether the event records have a matching page_views record.<jupyter_code>%time page_views.filter("uuid = 'a34004004c3e50' and document_id = 140264").show()
%time events.filter("uuid = 'a34004004c3e50' and document_id = 140264").show()<jupyter_output>+----------+--------------+-----------+-------------------+--------+------------+
|display_id|          uuid|document_id|          timestamp|platform|geo_location|
+----------+--------------+-----------+-------------------+--------+------------+
|      1341|a34004004c3e50|     140264|2016-06-14 09:31:33|       1|   US>CA>868|
|  17427544|a34004004c3e50|     140264|2016-06-16 19:24:19|       1|   US>CA>868|
+----------+--------------+-----------+-------------------+--------+------------+

CPU times: user 0 ns, sys: 0 ns, total: 0 ns
Wall time: 1.34 s<jupyter_text>Hypothesis: The page views record come web server logs. Events some user tracking devices such as omniture, google analytics. User may open the page, the after sometime, user may choose to view an ad.  User may view a page more than once. Find how many users have viewed the same page more than once.<jupyter_code>repeated_page_views = page_views.groupBy(["uuid", "document_id"]).count()\
        .filter("count > 1").orderBy(F.desc("count"))
repeated_page_views.show()
repeated_page_views.count()<jupyter_output><empty_output><jupyter_text>Look at a sample uuid who has repeated visited a page to observe the pattern in the source, location and timestamp<jupyter_code>if repeated_page_views.count()>0:
    sample_record = repeated_page_views.sample(True, 0.1).take(1)[0]
    page_views.filter(F.col("uuid") == sample_record.uuid)\
        .filter(F.col("document_id") == sample_record.document_id).show()<jupyter_output><empty_output><jupyter_text>Advertisement (Promopoted Content)<jupyter_code>promoted_contents = load("promoted_content")
promoted_contents.show()
promoted_contents.count()<jupyter_output><empty_output><jupyter_text>Promoted_content stores the meta of the ads. Double check the ad_id is unique in this dataset.<jupyter_code>%time promoted_contents.groupBy("ad_id").count().filter("count>1").count()<jupyter_output>CPU times: user 0 ns, sys: 0 ns, total: 0 ns
Wall time: 135 ms<jupyter_text>How many unique campaigns, documents and advertisers are there?<jupyter_code>%time promoted_contents\
.selectExpr("count(distinct document_id)", "count(distinct campaign_id)", "count(distinct advertiser_id)").first()<jupyter_output>CPU times: user 4 ms, sys: 0 ns, total: 4 ms
Wall time: 214 ms<jupyter_text>Does ad_id in the clicks dataset have meta data info in promoted_contents?<jupyter_code>%time (clicks_train.select("ad_id").union(clicks_test.select("ad_id"))\
       .join(promoted_contents, on = ["ad_id"], how = "leftanti").count())<jupyter_output>CPU times: user 0 ns, sys: 4 ms, total: 4 ms
Wall time: 3.71 s<jupyter_text>So, all the ad_id in clicks dataset exist in the promoted_contents.  Find average CTR for campaign<jupyter_code>avg_ctr_by_campaign = promoted_contents.join(ctrs, on = "ad_id").groupBy("campaign_id")\
.agg(F.avg("ctr").alias("avg_ctr"))

avg_ctr_by_campaign.show()<jupyter_output>+-----------+-------------------+
|campaign_id|            avg_ctr|
+-----------+-------------------+
|      12020|0.18853094256899317|
|      15551|0.14168937329700274|
|      16636|0.06654206156723358|
|      17774| 0.4740740740740741|
|      10710|0.07173178014624783|
|       7857|0.15448798511873105|
|       8748|   0.13014925725663|
|      28587|0.03608090199020851|
|      26563|0.05487204724409449|
|      25870|0.09421250207776988|
|      12787|0.26873419406362303|
|      24189|0.12057582351582147|
|        343| 0.1642336045511124|
|      22274|0.14866234594342462|
|      27610|0.12141843236348225|
|      25673| 0.0909153570443893|
|        692|0.10861946684564693|
|      17645|0.12021702251393619|
|      23815|  0.085025248657077|
|      26133| 0.1047878333543433|
+-----------+-------------------+
only showing top 20 rows<jupyter_text>Find avg ctr by advertiser.<jupyter_code>avg_ctr_by_advertiser = promoted_contents.join(ctrs, on = "ad_id").groupBy("advertiser_id")\
.agg(F.avg("ctr").alias("avg_ctr"))

avg_ctr_by_advertiser.show()<jupyter_output>+-------------+-------------------+
|advertiser_id|            avg_ctr|
+-------------+-------------------+
|         2349|0.10883058103462902|
|         3829| 0.2068642691859514|
|         1283|0.14176344970802937|
|          434| 0.1043910399182465|
|         2030|0.10217395554446719|
|         2447|0.05872570281791586|
|         2625|0.13082804384293295|
|           38|0.13581588111063284|
|         1633|0.20279203966409717|
|         3404|0.03207982667240285|
|         1809|0.15295881650970958|
|         2260|0.35096740188296494|
|          881| 0.1252447139156763|
|         1557|0.09639892315222712|
|         1830| 0.2122485949912904|
|         2116| 0.1282686945248045|
|         2840|0.23191125858194017|
|         3572|0.12020942448693174|
|          784|0.08288804563485187|
|         2859|0.19682314282065358|
+-------------+-------------------+
only showing top 20 rows<jupyter_text>Find avg ctr by document.<jupyter_code>avg_ctr_by_document = promoted_contents.join(ctrs, on = "ad_id").groupBy("document_id")\
.agg(F.avg("ctr").alias("avg_ctr"))

avg_ctr_by_document.show()<jupyter_output>+-----------+--------------------+
|document_id|             avg_ctr|
+-----------+--------------------+
|    1087952| 0.33327873327873325|
|     870875| 0.24400340797438524|
|    1377696|0.048441767634379414|
|    1146963| 0.19168084171412803|
|    1083829| 0.32597524434580355|
|     378777| 0.08661417322834646|
|    1247211|  0.2608837247293875|
|     902349|0.058823529411764705|
|     907267| 0.03737655352703233|
|    1273596| 0.26410511426215816|
|    1333191|  0.1767042599220612|
|     876083| 0.08421440203631031|
|    1767220| 0.16518679146475518|
|    1603476| 0.19174833310958966|
|    1561030| 0.05284776441477035|
|    1176283|  0.3650056262723164|
|     630595|0.014084507042253521|
|    1590983|                 0.0|
|    1154277| 0.19185631204008377|
|    1326394|  0.0379591665969375|
+-----------+--------------------+
only showing top 20 rows<jupyter_text>Document Attributes Document Meta Data<jupyter_code>documents_meta = load("documents_meta")
documents_meta.show(5, False)
documents_meta.count()<jupyter_output><empty_output><jupyter_text>Verify whether the document_id is unique in document_meta.<jupyter_code>documents_meta.groupby("document_id").count().filter("count>1").count()<jupyter_output><empty_output><jupyter_text>How many source_ids are there?<jupyter_code>documents_meta.select("source_id").distinct().count()<jupyter_output><empty_output><jupyter_text>How many publisher_ids are there?<jupyter_code>documents_meta.select("publisher_id").distinct().count()<jupyter_output><empty_output><jupyter_text>Document Categories<jupyter_code>documents_categories = load("documents_categories").drop_duplicates(["document_id", "category_id"])
documents_categories.printSchema()
documents_categories.show(5, False)
documents_categories.count()
documents_categories.select("category_id").distinct().count()<jupyter_output><empty_output><jupyter_text>category_id is not indexed. Using string indexer to index it. One Hot Encoding is another option. Here I am not doing it because the I want to use the confidence_level to weigh in.<jupyter_code>from pyspark.ml.feature import StringIndexer

#String indexer required string column
documents_categories = documents_categories.withColumn("category_id", F.expr("cast(category_id as string)")) 

if "id" in documents_categories.columns:
    documents_categories = documents_categories.drop("id")

category_indexer = StringIndexer(inputCol="category_id", outputCol="id")
documents_categories = category_indexer.fit(documents_categories).transform(documents_categories)

documents_categories.show()<jupyter_output>+-----------+-----------+----------------+----+
|document_id|category_id|confidence_level|  id|
+-----------+-----------+----------------+----+
|    2416431|       1407|      0.87841876|12.0|
|    2260393|       1702|            0.92| 1.0|
|     917692|       1702|            0.92| 1.0|
|    1786173|       1706|            0.07|13.0|
|    1783554|       1709|            0.07|55.0|
|    1776902|       1806|     0.597394548|20.0|
|    1784492|       1702|     0.556040824| 1.0|
|    1812275|       1903|            0.92|15.0|
|    1762888|       1613|     0.904981736|38.0|
|    1811648|       1702|            0.92| 1.0|
|    1110915|       1702|     0.622750116| 1.0|
|     983795|       1702|     0.582905652| 1.0|
|    1093318|       2004|            0.07| 7.0|
|     984200|       1302|     0.436507432|48.0|
|    1913424|       1706|            0.07|13.0|
|    2061034|       1702|     0.042858662| 1.0|
|    1972828|       1702|            0.92| 1.0|
|    1904284|       1706|            0.0[...]<jupyter_text>Group the data by document_id and pack the other information in a array field.<jupyter_code>documents_categories_n = documents_categories\
.withColumn("pair", F.struct("id", "confidence_level"))\
.groupBy("document_id")\
.agg(F.collect_list("pair").alias("categories"))

documents_categories_n.printSchema()
documents_categories_n.count()
documents_categories_n.show(3, False)<jupyter_output>+-----------+-----------------------------------------+
|document_id|categories                               |
+-----------+-----------------------------------------+
|12         |[[0.0, 0.024439128], [15.0, 0.321199968]]|
|18         |[[12.0, 0.92], [30.0, 0.07]]             |
|38         |[[51.0, 0.92], [32.0, 0.07]]             |
+-----------+-----------------------------------------+
only showing top 3 rows<jupyter_text>Document entities<jupyter_code>documents_entities = load("documents_entities").drop_duplicates(["document_id", "entity_id"])
documents_entities.show(5, False)<jupyter_output>Number of partitions for df documents_entities: 8
+-----------+--------------------------------+-----------------+
|document_id|entity_id                       |confidence_level |
+-----------+--------------------------------+-----------------+
|1090777    |13382f1237391289a06215b1c3fd76b6|0.242424004409274|
|1118903    |135629f3dea09fb3793708b55f49ed41|0.327493886222015|
|871008     |00eb36948f331eada0a7306328e8cf9b|0.683590151896937|
|2779469    |a1729c4252069e909f56b4c41283495b|0.248229877540157|
|2126240    |8eff7a6a560c2fd02b0bf53db50ef3c5|0.204488577325427|
+-----------+--------------------------------+-----------------+
only showing top 5 rows<jupyter_text>Find stats around the number of entities per document.<jupyter_code>documents_entities.groupBy("document_id").count().select("count").describe().show()<jupyter_output>+-------+------------------+
|summary|             count|
+-------+------------------+
|  count|           1791420|
|   mean| 3.091152270266046|
| stddev|2.3904599699254128|
|    min|                 1|
|    max|                10|
+-------+------------------+<jupyter_text>Find count of unique entity_ids.<jupyter_code>documents_entities.select("entity_id").distinct().count()<jupyter_output><empty_output><jupyter_text>Apply StringIndex to index the entity_id.<jupyter_code>if "id" in documents_entities.columns:
    documents_entities = documents_entities.drop("id")

documents_entities =  StringIndexer(inputCol="entity_id", outputCol="id")\
.fit(documents_entities)\
.transform(documents_entities)

documents_entities.show()
documents_entities_n = documents_entities\
.withColumn("pair", F.struct("id", "confidence_level"))\
.groupBy("document_id")\
.agg(F.collect_list("pair").alias("entities"))

documents_entities_n.printSchema()<jupyter_output>root
 |-- document_id: integer (nullable = true)
 |-- entities: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- id: double (nullable = false)
 |    |    |-- confidence_level: double (nullable = true)<jupyter_text>Document topics<jupyter_code>documents_topics = load("documents_topics").drop_duplicates(["document_id","topic_id"])
documents_topics.show(5, False)
print(sorted([v.topic_id for v in documents_topics.select("topic_id").distinct().collect()]))<jupyter_output>[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99, 100, 101, 102, 103, 104, 105, 106, 107, 108, 109, 110, 111, 112, 113, 114, 115, 116, 117, 118, 119, 120, 121, 122, 123, 124, 125, 126, 127, 128, 129, 130, 131, 132, 133, 134, 135, 136, 137, 138, 139, 140, 141, 142, 143, 144, 145, 146, 147, 148, 149, 150, 151, 152, 153, 154, 155, 156, 157, 158, 159, 160, 161, 162, 163, 164, 165, 166, 167, 168, 169, 170, 171, 172, 173, 174, 175, 176, 177, 178, 179, 180, 181, 182, 183, 184, 185, 186, 187, 188, 189, 190, 191, 192, 193, 194, 195, 196, 197, 198, 199, 200, 201, 202, 203, 204, 205, 206, 207, 208, 209, 210, 211, 212, 213, 214, 215, 216, 217, 218, 219, 220, 221,[...]<jupyter_text>Topic_id seems already indexed. And there 300 topics are there.<jupyter_code>documents_topics.select("topic_id").distinct().count()
documents_topics_n = documents_topics\
.toDF("document_id", "id", "confidence_level")\
.withColumn("pair", F.struct("id", "confidence_level"))\
.groupBy("document_id")\
.agg(F.collect_list("pair").alias("topics"))

documents_topics_n.printSchema()
documents_topics_n.count()<jupyter_output><empty_output><jupyter_text>Create a field new field for each dataset to indicate the original source file and then join all 4 datasets - categories, entities, topics and meta<jupyter_code>docs = documents_meta.join(documents_categories_n, on = "document_id", how = "full")
docs = docs.join(documents_topics_n, on = "document_id", how = "full")
docs = docs.join(documents_entities_n, on = "document_id", how = "full")
docs.persist(StorageLevel.DISK_ONLY).count()
docs.printSchema()
%show docs<jupyter_output><empty_output><jupyter_text>Find count of null values of each type of information - meta, category, entity and topic.<jupyter_code>docs.selectExpr("sum(if(isnull(topics), 1, 0)) null_topics"
              , "sum(if(isnull(categories), 1, 0)) null_categories"
              , "sum(if(isnull(entities), 1, 0)) null_entities"
              , "sum(if(isnull(publisher_id), 1, 0)) null_meta").first()<jupyter_output><empty_output><jupyter_text>Vectorize and caculate weighted IDF scores How many document topics are there?<jupyter_code>docs.select(F.explode("topics")).select("col.id").distinct().count()<jupyter_output><empty_output><jupyter_text>How many document categories are there?<jupyter_code>docs.select(F.explode("categories")).select("col.id").distinct().count()<jupyter_output><empty_output><jupyter_text>How many document entities are there?<jupyter_code>docs.select(F.explode("entities")).select("col.id").distinct().count()
from pyspark.ml.linalg import SparseVector, VectorUDT

def to_vector(values, n):
    if values is not None:
        values = sorted(values, key=lambda v: v.id)
        indices = [v.id for v in values]
        values = [v.confidence_level for v in values]
        return SparseVector(n, indices, values)
    return SparseVector(n, [], [])

spark.udf.register("to_vector", to_vector, VectorUDT())

docs_vectorized = docs\
.withColumn("topics_vector", F.expr("to_vector(topics, 300)"))\
.withColumn("categories_vector", F.expr("to_vector(categories, 97)"))
#.withColumn("entities_vector", F.expr("to_vector(entities, 1326009)"))

docs_vectorized.printSchema()
docs_vectorized.cache().count()
docs_vectorized.select("topics_vector").first()
docs_vectorized.select("categories_vector").count()
from pyspark.ml.feature import IDF, Tokenizer

if "topics_idf" in docs.columns:
    docs = docs.drop("topics_idf")
if "entities_idf" in docs.columns:
    docs = docs.drop("entities_idf")
if "categories_idf" in docs.columns:
    docs = docs.drop("categories_idf")

topics_idf = IDF(inputCol="topics_vector", outputCol="topics_idf")
entities_idf = IDF(inputCol="entities_vector", outputCol="entities_idf")
categories_idf = IDF(inputCol="categories_vector", outputCol="categories_idf")

df1 = docs_vectorized
df2 = topics_idf.fit(df1).transform(df1).cache()
df3 = categories_idf.fit(df2).transform(df2).cache()
#df4 = entities_idf.fit(df3).transform(df3).cache()

docs_idf = df3
docs_idf.printSchema()
docs_idf.select("document_id", "topics_idf", "categories_idf").first()<jupyter_output><empty_output><jupyter_text>Feature Generation  User profile| Column  |  Description ||---|---||user_has_already_viewed_doc|  For each content recommended to the user, verify whether the user had previously visited that pages.  |user_views_count | Do eager readers behave differently from other users? Let’s add this feature and let machine learning models guess that.|user_views_categories, user_views_topics, user_views_entities | User profile vectors based on categories, topics and entities of documents that users have previously viewed (weighted by confidence and TF-IDF), to model users preferences in a Content-Based Filtering approach|user_avg_views_of_distinct_docs | Ratio between (user_distinct_docs_views / user_views), indicating how often users read previously visited pages again.<jupyter_code>user_has_already_viewed_doc = (page_views
.withColumn("user_has_already_viewed_doc"
            , F.expr("((ROW_NUMBER() OVER (PARTITION BY uuid, document_id ORDER BY timestamp))) > 1"))
.select("uuid", "document_id", "timestamp", "user_has_already_viewed_doc")
             )
%show user_has_already_viewed_doc.filter("uuid = '6c4a7527da27d7' and document_id = 38922")
user_views_count = (page_views
    .withColumn("user_views_count", 
        F.expr("COUNT(1) OVER (PARTITION BY uuid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND -1 FOLLOWING)"))
    .select("uuid", "timestamp", "user_views_count"))

%show user_views_count.filter("uuid = '6c4a7527da27d7' and document_id = 38922")
#page_views = page_views.withColumn("user_avg_views_of_distinct_docs", F.expr("COUNT(distinct document_id) " + 
#                "OVER (PARTITION BY uuid ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND -1 FOLLOWING)"))
#
#%show page_views.filter("uuid = '6c4a7527da27d7' and document_id = 38922")<jupyter_output><empty_output><jupyter_text>Ads and Documents| Column  |  Description ||---|---||doc_ad_days_since_published, doc_event_days_since_published | Days elapsed since the ad document was published in a given user visit. The general assumption is that new content is more relevant to users. But if you are reading an old post, you might be interested in other old posts.|doc_avg_views_by_distinct_users_cf | Average page views of the ad document by distinct users. Is this a webpage people usually return to?|ad_views_count, doc_views_count|How popular is a document or ad?<jupyter_code>doc_event_days_since_published = (events
    .join(documents_meta, on = "document_id")
    .selectExpr("display_id"
                , "document_id"
                , "timestamp"
                , "publish_time"
                , "datediff(timestamp, publish_time) age")
)

doc_event_days_since_published.show()
page_view_count_by_document_id = page_views.groupBy("document_id")\
.count().withColumn("page_view_count_by_document_id", F.expr("log(count)"))\
.select("document_id", "page_view_count_by_document_id")

page_view_count_by_document_id.show()
page_view_count_by_uuid = (page_views
.groupBy("uuid")
.count()
.withColumn("page_view_count_by_uuid", F.expr("log(count)"))
.select("uuid", "page_view_count_by_uuid"))

page_view_count_by_uuid.show()<jupyter_output>+--------------+-----------------------+
|          uuid|page_view_count_by_uuid|
+--------------+-----------------------+
|6ec2f1df169162|                    0.0|
|682ff7de2c98c2|                    0.0|
|bc12ea5e483fc0|                    0.0|
|df0e90379a76d6|                    0.0|
|3286c54a862512|     0.6931471805599453|
|fa69933c24d064|                    0.0|
|329a791d36dedb|                    0.0|
|3cd95a6e6271a6|                    0.0|
|ab642647e0361a|                    0.0|
|7f6657ac2a45c1|                    0.0|
|5c61bf0cedf30f|                    0.0|
|c9b5daf2b577d0|                    0.0|
|e870b628cf651b|                    0.0|
|6be4e8057c3c71|                    0.0|
|a4d1897a905a05|                    0.0|
|27e48f2ec0626a|                    0.0|
|6f3a61ca09e99a|                    0.0|
|b5789a234c82a2|     0.6931471805599453|
|5131b2d9a6c741|                    0.0|
|d77ffd5f6727a9|                    0.0|
+--------------+-----------------------+
only showing top[...]<jupyter_text>Events| Column  |  Description ||---|---||event_local_hour (binned), event_weekend | Event timestamps were in UTC-4, so I processed event geolocation to get timezones and adjust for users' local time. They were binned in periods like morning, afternoon, midday, evening, night. A flag indicating whether it was a weekend was also included. The assumption here is that time influences the kind of content users will read.|event_country, event_country_state | The field event_geolocation was parsed to extract the user’s country and state in a page visit.|ad_id, doc_event_id, doc_ad_id, ad_advertiser, … | All of the original categorical fields were One-Hot Encoded to be used by the models, generating about 126,000 features.<jupyter_code>events.selectExpr("uuid", "geo_location"
                  , "split(geo_location, '>')[0] country"
                  , "split(geo_location, '>')[1] state"
                 ).show()
events.selectExpr("split(geo_location, '>')[0] country").distinct().toPandas()
(events
 .selectExpr("display_id", "timestamp", "hour(timestamp) hour")
 .withColumn("day_session", F.expr("hour % 8"))).show()<jupyter_output>+----------+-------------------+----+-----------+
|display_id|          timestamp|hour|day_session|
+----------+-------------------+----+-----------+
|         1|2016-06-14 09:30:00|   9|          1|
|         2|2016-06-14 09:30:00|   9|          1|
|         3|2016-06-14 09:30:00|   9|          1|
|         4|2016-06-14 09:30:00|   9|          1|
|         5|2016-06-14 09:30:00|   9|          1|
|         6|2016-06-14 09:30:00|   9|          1|
|         7|2016-06-14 09:30:00|   9|          1|
|         8|2016-06-14 09:30:00|   9|          1|
|         9|2016-06-14 09:30:00|   9|          1|
|        10|2016-06-14 09:30:00|   9|          1|
|        11|2016-06-14 09:30:00|   9|          1|
|        12|2016-06-14 09:30:00|   9|          1|
|        13|2016-06-14 09:30:00|   9|          1|
|        14|2016-06-14 09:30:00|   9|          1|
|        15|2016-06-14 09:30:00|   9|          1|
|        16|2016-06-14 09:30:01|   9|          1|
|        17|2016-06-14 09:30:01|   9|          1|
[...]<jupyter_text>Top countries by ad clicks<jupyter_code>events.selectExpr("split(geo_location, '>')[0] country")\
.groupBy("country").count().orderBy(F.desc("count")).show()<jupyter_output>+-------+--------+
|country|   count|
+-------+--------+
|     US|18595452|
|     CA| 1215350|
|     GB| 1117544|
|     AU|  483021|
|     IN|  228461|
|     ZA|  111523|
|     NZ|  109802|
|     PH|   85338|
|     DE|   82384|
|     SG|   81975|
|     MY|   53398|
|     NL|   51209|
|     NG|   41946|
|     IE|   39505|
|     SE|   38931|
|     FR|   38755|
|     MX|   35044|
|     IT|   28727|
|     KE|   26607|
|     JP|   25955|
+-------+--------+
only showing top 20 rows<jupyter_text>Average CTR| Column  |  Description ||---|---||avg_ctr_ad_id, avg_ctr_publisher_id, avg_ctr_advertiser_id, avg_ctr_campain_id, avg_ctr_entity_id_country … | Average CTR (clicks / views) given some categorical combinations and CTR confidence (details on Part II post). Eg. P(click  category01, category02).<jupyter_code>events.cache()
clicks_train.cache()
documents_meta.cache()
promoted_contents.cache()
avg_ctrs_by_ad_id = clicks_train.groupBy("ad_id").agg(F.avg("clicked").alias("avg_ctr_by_ad_id"))
%show avg_ctrs_by_ad_id
avg_ctrs_by_campaign_id = (clicks_train
.join(promoted_contents, on = "ad_id")
.groupBy("campaign_id")
.agg(F.avg("clicked").alias("avg_ctr_by_campaign_id")))

%show avg_ctrs_by_campaign_id
avg_ctrs_by_advertiser_id = (clicks_train
.join(promoted_contents, on = "ad_id")
.groupBy("advertiser_id")
.agg(F.avg("clicked").alias("avg_ctr_by_advertiser_id"))
.cache()
)

%show avg_ctrs_by_advertiser_id
avg_ctrs_by_document_id = (clicks_train
.join(promoted_contents, on = "ad_id")
.groupBy("document_id")
.agg(F.avg("clicked").alias("avg_ctr_by_document_id"))
.cache()
)

%show avg_ctrs_by_document_id
avg_ctrs_by_time = (events
 .join(documents_meta, on = "document_id", how = "left")
 .join(clicks_train, on = "display_id", how = "left")
 .join(promoted_contents, on = "ad_id", how = "left")
 .withColumn("total_clicks_by_ad_id"
    , F.expr("SUM(clicked) OVER (PARTITION BY ad_id ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND -1 FOLLOWING)"))
 .withColumn("total_events_by_ad_id"
    , F.expr("COUNT(*) OVER (PARTITION BY ad_id ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND -1 FOLLOWING)"))

 .withColumn("total_clicks_by_advertiser_id"
    , F.expr("SUM(clicked) OVER (PARTITION BY advertiser_id ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND -1 FOLLOWING)"))
 .withColumn("total_events_by_advertiser_id"
    , F.expr("COUNT(*) OVER (PARTITION BY advertiser_id ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND -1 FOLLOWING)"))

 .withColumn("total_clicks_by_campaign_id"
    , F.expr("SUM(clicked) OVER (PARTITION BY campaign_id ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND -1 FOLLOWING)"))
 .withColumn("total_events_by_campaign_id"
    , F.expr("COUNT(*) OVER (PARTITION BY campaign_id ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND -1 FOLLOWING)"))

 .withColumn("total_clicks_by_publisher_id"
    , F.expr("SUM(clicked) OVER (PARTITION BY publisher_id ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND -1 FOLLOWING)"))
 .withColumn("total_events_by_publisher_id"
    , F.expr("COUNT(*) OVER (PARTITION BY publisher_id ORDER BY timestamp ROWS BETWEEN UNBOUNDED PRECEDING AND -1 FOLLOWING)"))
 .selectExpr("display_id"
             , "timestamp"
             , "(total_clicks_by_advertiser_id/total_events_by_advertiser_id) avg_ctr_by_advertiser_id"
             , "(total_clicks_by_campaign_id/total_events_by_campaign_id) avg_ctr_by_campaign_id"
             , "(total_clicks_by_ad_id/total_events_by_ad_id) avg_ctr_by_ad_id"
             , "(total_clicks_by_publisher_id/total_events_by_publisher_id) avg_ctr_by_publisher_id")
)

%show avg_ctrs_by_time<jupyter_output><empty_output><jupyter_text>Content-Based Similarities| Column  |  Description ||---|---||user_doc_ad_sim_categories, user_doc_ad_sim_topics, user_doc_ad_sim_entities | Cosine similarity between user profile and ad document profile vectors (TF-IDF).|doc_event_doc_ad_sim_categories, doc_event_doc_ad_sim_topics, doc_event_doc_ad_sim_entities | Cosine similarity between event document (landing page context) and ad document profile vectors (TF-IDF).  Prepare training and test datasets<jupyter_code>docs.printSchema()
docs_idf.drop("document_id").printSchema()
clicks = (clicks_train.withColumn("is_train", F.lit(1))
            .union(clicks_test
                   .withColumn("clicked", F.lit(0))
                   .withColumn("is_train", F.lit(0))))

df = (clicks
            .join(events.alias("events"), on = ["display_id"], how = "left")
            .join(docs_idf.alias("docs_idf"), on = ["document_id"], how = "left")
            .join(promoted_contents.drop("document_id"), on = ["ad_id"], how = "left")
            .join(page_view_count_by_uuid, on = ["uuid"], how = "left")
            .join(page_view_count_by_document_id, on = ["document_id"], how = "left")
      
            .withColumn("clicks_by_ad_id", F.expr("sum(clicked) over (partition by ad_id)"))
            .withColumn("events_by_ad_id", F.expr("count(*) over (partition by ad_id)"))
            .withColumn("avg_ctr_by_ad_id", F.expr("clicks_by_ad_id/events_by_ad_id"))

            .withColumn("clicks_by_campaign_id", F.expr("sum(clicked) over (partition by campaign_id)"))
            .withColumn("events_by_campaign_id", F.expr("count(*) over (partition by campaign_id)"))
            .withColumn("avg_ctr_by_campaign_id", F.expr("clicks_by_campaign_id/events_by_campaign_id"))
            
            .withColumn("clicks_by_document_id", F.expr("sum(clicked) over (partition by events.document_id)"))
            .withColumn("events_by_document_id", F.expr("count(*) over (partition by events.document_id)"))
            .withColumn("avg_ctr_by_document_id", F.expr("clicks_by_campaign_id/events_by_document_id"))
            
            .withColumn("clicks_by_advertiser_id", F.expr("sum(clicked) over (partition by advertiser_id)"))
            .withColumn("events_by_advertiser_id", F.expr("count(*) over (partition by advertiser_id)"))
            .withColumn("avg_ctr_by_advertiser_id", F.expr("clicks_by_campaign_id/events_by_advertiser_id"))
            
            .withColumn("country", F.expr("split(geo_location, '>')[0]"))
            .withColumn("state", F.expr("split(geo_location, '>')[1]"))
            .withColumn("doc_age", F.expr("datediff(timestamp, publish_time)"))
            .withColumn("session", F.expr("cast((hour(timestamp) % 8) as string)"))
            .withColumn("source_id", F.expr("cast(source_id as string)"))
            .withColumn("publisher_id", F.expr("cast(publisher_id as string)"))
           )
df.printSchema()
df.write.mode("overwrite").save(base_path + "merged_enriched")
df.printSchema()<jupyter_output>root
 |-- document_id: integer (nullable = true)
 |-- uuid: string (nullable = true)
 |-- ad_id: integer (nullable = true)
 |-- display_id: integer (nullable = true)
 |-- clicked: integer (nullable = true)
 |-- is_train: integer (nullable = false)
 |-- timestamp: timestamp (nullable = true)
 |-- platform: string (nullable = true)
 |-- geo_location: string (nullable = true)
 |-- source_id: string (nullable = true)
 |-- publisher_id: string (nullable = true)
 |-- publish_time: timestamp (nullable = true)
 |-- categories: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- id: double (nullable = false)
 |    |    |-- confidence_level: double (nullable = true)
 |-- topics: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- id: integer (nullable = true)
 |    |    |-- confidence_level: double (nullable = true)
 |-- entities: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- id: double[...]<jupyter_text>Machine Learning<jupyter_code>df = load("merged_enriched", cache = False)
df.count()
features = [
   'platform'
 , 'source_id'
 , 'publisher_id'
 , 'topics_idf'
 , 'categories_idf'
 , 'avg_ctr_by_ad_id'
 , 'avg_ctr_by_campaign_id'
 , 'avg_ctr_by_document_id'
 , 'avg_ctr_by_advertiser_id'
 , "country"
 , "state"
 , "doc_age"
 , "session"
 , "ad_id"
 , "display_id"
 , "is_train"
 , "clicked"
]
df.selectExpr(*features).printSchema()
%show df
df1 = df<jupyter_output><empty_output><jupyter_text>Detect and impute null values<jupyter_code>def show_null_counts(df):
    null_testers = ["sum(if(isnull(%s), 1, 0)) %s" % (f, f) for f in df.columns]
    null_counts = df.selectExpr(*null_testers).toPandas().T
    null_counts.columns = ["Count"]
    null_counts["pct"] = null_counts.Count/df.count()
    null_counts["dtype"] = [t[1] for t in df.dtypes]
    print(null_counts.to_string())

show_null_counts(df.selectExpr(*features))
df_trunc = df.selectExpr(*features)
distinct_counts = df_trunc.selectExpr(*["approx_count_distinct(%s)"
                        % f for f in df.selectExpr(*features).columns]).toPandas()
print(distinct_counts.T.to_string())
fill_na_values = {"platform": "<null>"
    , "source_id": "<null>"
    , "publisher_id": "<null>"
    , "avg_ctr_by_ad_id": 0.0
    , "avg_ctr_by_campaign_id": 0.0
    , "avg_ctr_by_document_id": 0.0
    , "avg_ctr_by_advertiser_id": 0.0
    , "country": "null"
    , "state": "null"
    , "doc_age": -1}

df_null_removed = df.selectExpr(*features).na.fill(fill_na_values)
show_null_counts(df_null_removed)<jupyter_output>Count  pct   dtype
platform                      0  0.0  string
source_id                     0  0.0  string
publisher_id                  0  0.0  string
topics_idf                    0  0.0  vector
categories_idf                0  0.0  vector
avg_ctr_by_ad_id              0  0.0  double
avg_ctr_by_campaign_id        0  0.0  double
avg_ctr_by_document_id        0  0.0  double
avg_ctr_by_advertiser_id      0  0.0  double
country                       0  0.0  string
state                         0  0.0  string
doc_age                       0  0.0     int
session                       0  0.0  string
ad_id                         0  0.0     int
display_id                    0  0.0     int
is_train                      0  0.0     int
clicked                       0  0.0     int<jupyter_text>Apply StringIndexing and OneHotE Convert categorical values (string type) into index values, subsequently to one hot encoded values<jupyter_code>from pyspark.ml.feature import OneHotEncoderEstimator, StringIndexer, VectorAssembler
categorical_columns = [col for col, dtype in df_null_removed.dtypes if dtype == "string"]

df_string_indexed = df_null_removed
for col in categorical_columns:
    indexer = StringIndexer(inputCol=col, outputCol="%s_index" % col)
    df_string_indexed = indexer.fit(df_string_indexed).transform(df_string_indexed)

one_hot_estimator = OneHotEncoderEstimator(
    inputCols = [col + "_index" for col in categorical_columns], 
    outputCols = [col + "_vec" for col in categorical_columns]
)
df_ohe = one_hot_estimator.fit(df_string_indexed).transform(df_string_indexed)
df_ohe.dtypes
to_be_vectorized = [('topics_idf', 'vector'),
 ('categories_idf', 'vector'),
 ('avg_ctr_by_ad_id', 'double'),
 ('avg_ctr_by_campaign_id', 'double'),
 ('avg_ctr_by_document_id', 'double'),
 ('avg_ctr_by_advertiser_id', 'double'),
 ('doc_age', 'int'),
 ('country_vec', 'vector'),
 ('session_vec', 'vector'),
 ('source_id_vec', 'vector'),
 ('state_vec', 'vector'),
 ('publisher_id_vec', 'vector'),
 ('platform_vec', 'vector')]

vector_assembler = VectorAssembler(inputCols = [c for c, _ in to_be_vectorized], outputCol="features")
df_vectorized = vector_assembler.transform(df_ohe)
df_vectorized.dtypes
df_train, df_test = df_vectorized.filter("is_train = 1").select("display_id", "ad_id", "clicked", "features")\
.randomSplit(weights=[0.7, 0.3], seed = 1)

cache_df(df_train, "df_train")
df_train.printSchema()
df_train.count()
from pyspark.ml.classification import LogisticRegression
lr = LogisticRegression(maxIter=10, regParam=0.1, elasticNetParam=0.8, featuresCol="features", labelCol="clicked")

lrModel = lr.fit(df_train)

print("Coefficients: " + str(lrModel.coefficients))
print("Intercept: " + str(lrModel.intercept))

trainingSummary = lrModel.summary
objectiveHistory = trainingSummary.objectiveHistory
print("objectiveHistory:")
for objective in objectiveHistory:
    print(objective)

# Obtain the receiver-operating characteristic as a dataframe and areaUnderROC.
trainingSummary.roc.show()
print("areaUnderROC: " + str(trainingSummary.areaUnderROC))
# for multiclass, we can inspect metrics on a per-label basis
print("False positive rate by label:")
for i, rate in enumerate(trainingSummary.falsePositiveRateByLabel):
    print("label %d: %s" % (i, rate))

print("True positive rate by label:")
for i, rate in enumerate(trainingSummary.truePositiveRateByLabel):
    print("label %d: %s" % (i, rate))

print("Precision by label:")
for i, prec in enumerate(trainingSummary.precisionByLabel):
    print("label %d: %s" % (i, prec))

print("Recall by label:")
for i, rec in enumerate(trainingSummary.recallByLabel):
    print("label %d: %s" % (i, rec))

print("F-measure by label:")
for i, f in enumerate(trainingSummary.fMeasureByLabel()):
    print("label %d: %s" % (i, f))

accuracy = trainingSummary.accuracy
falsePositiveRate = trainingSummary.weightedFalsePositiveRate
truePositiveRate = trainingSummary.weightedTruePositiveRate
fMeasure = trainingSummary.weightedFMeasure()
precision = trainingSummary.weightedPrecision
recall = trainingSummary.weightedRecall
print("Accuracy: %s\nFPR: %s\nTPR: %s\nF-measure: %s\nPrecision: %s\nRecall: %s"
      % (accuracy, falsePositiveRate, truePositiveRate, fMeasure, precision, recall))
lrModel.write().overwrite().save(base_path + "lrModel")
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
# Make predictions.
predictions = lrModel.transform(df_test)

# Select example rows to display.
predictions.select("prediction", "clicked", "features").show(5)

# Select (prediction, true label) and compute test error
evaluator = MulticlassClassificationEvaluator(
    labelCol="clicked", predictionCol="prediction", metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print("Accuracy = %g " % (accuracy))<jupyter_output>+----------+-------+--------------------+
|prediction|clicked|            features|
+----------+-------+--------------------+
|       0.0|      1|(6487,[20,97,241,...|
|       0.0|      0|(6487,[8,56,93,10...|
|       0.0|      0|(6487,[183,199,30...|
|       0.0|      0|(6487,[71,199,229...|
|       0.0|      0|(6487,[16,29,65,1...|
+----------+-------+--------------------+
only showing top 5 rows
{'fraction_non_alphanumeric': 0.16566824644549763, 'fraction_numerical': 0.08585781990521327, 'mean_word_length': 3.4902110997616616, 'pattern_counts': {'":': 10, '<': 246, '<?xml version=': 0, '>': 277, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 45, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '70', 'dir': 'jupyter-structured-clean-dedup', 'id': '573441', 'license': 'Apache-2.0', 'n_tokens_mistral': 21274, 'n_tokens_neox': 18316, 'n_words': 3705, 'path': 'SparkML - 07 Click Prediction (Outbrain dataset).ipynb', 'repo_name': 'nishantm9/machine-learning', 'size': '52750'}