SDS-2.2, Scalable Data Science

Archived YouTube video of this live unedited lab-lecture:

Wiki Clickstream Analysis

** Dataset: 3.2 billion requests collected during the month of February 2015 grouped by (src, dest) **

** Source: https://datahub.io/dataset/wikipedia-clickstream/ **

This notebook requires Spark 1.6+.

This notebook was originally a data analysis workflow developed with Databricks Community Edition, a free version of Databricks designed for learning Apache Spark.

Here we elucidate the original python notebook (also linked here) used in the talk by Michael Armbrust at Spark Summit East February 2016 shared from https://twitter.com/michaelarmbrust/status/699969850475737088 (watch later)

Data set

The data we are exploring in this lab is the February 2015 English Wikipedia Clickstream data, and it is available here: http://datahub.io/dataset/wikipedia-clickstream/resource/be85cc68-d1e6-4134-804a-fd36b94dbb82.

According to Wikimedia:

"The data contains counts of (referer, resource) pairs extracted from the request logs of English Wikipedia. When a client requests a resource by following a link or performing a search, the URI of the webpage that linked to the resource is included with the request in an HTTP header called the "referer". This data captures 22 million (referer, resource) pairs from a total of 3.2 billion requests collected during the month of February 2015."

The data is approximately 1.2GB and it is hosted in the following Databricks file: /databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed

Let us first understand this Wikimedia data set a bit more

Let's read the datahub-hosted link https://datahub.io/dataset/wikipedia-clickstream in the embedding below. Also click the blog by Ellery Wulczyn, Data Scientist at The Wikimedia Foundation, to better understand how the data was generated (remember to Right-Click and use -> and <- if navigating within the embedded html frame below).

Run the next two cells for some housekeeping.

if (org.apache.spark.BuildInfo.sparkBranch < "1.6") sys.error("Attach this notebook to a cluster running Spark 1.6+")

Loading and Exploring the data

val data = sc.textFile("dbfs:///databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed")

data: org.apache.spark.rdd.RDD[String] = dbfs:///databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed MapPartitionsRDD[1] at textFile at <console>:34

Looking at the first few lines of the data

data.take(5).foreach(println)

prev_id    curr_id    n    prev_title    curr_title    type
    3632887    121    other-google    !!    other
    3632887    93    other-wikipedia    !!    other
    3632887    46    other-empty    !!    other
    3632887    10    other-other    !!    other

data.take(2)

res3: Array[String] = Array(prev_id    curr_id    n    prev_title    curr_title    type, "    3632887    121    other-google    !!    other")

• The first line looks like a header
• The second line (separated from the first by ",") contains data organized according to the header, i.e., prev_id = 3632887, curr_id = 121", and so on.

Actually, here is the meaning of each column:

• prev_id: if the referer does not correspond to an article in the main namespace of English Wikipedia, this value will be empty. Otherwise, it contains the unique MediaWiki page ID of the article corresponding to the referer i.e. the previous article the client was on

• curr_id: the MediaWiki unique page ID of the article the client requested

• prev_title: the result of mapping the referer URL to the fixed set of values described below

• curr_title: the title of the article the client requested

• n: the number of occurrences of the (referer, resource) pair

• type

  • "link" if the referer and request are both articles and the referer links to the request
  • "redlink" if the referer is an article and links to the request, but the request is not in the production enwiki.page table
  • "other" if the referer and request are both articles but the referer does not link to the request. This can happen when clients search or spoof their refer

Referers were mapped to a fixed set of values corresponding to internal traffic or external traffic from one of the top 5 global traffic sources to English Wikipedia, based on this scheme:

• an article in the main namespace of English Wikipedia -> the article title
• any Wikipedia page that is not in the main namespace of English Wikipedia -> other-wikipedia
• an empty referer -> other-empty
• a page from any other Wikimedia project -> other-internal
• Google -> other-google
• Yahoo -> other-yahoo
• Bing -> other-bing
• Facebook -> other-facebook
• Twitter -> other-twitter
• anything else -> other-other

In the second line of the file above, we can see there were 121 clicks from Google to the Wikipedia page on "!!" (double exclamation marks). People search for everything!

• prev_id = (nothing)
• curr_id = 3632887 --> (Wikipedia page ID)
• n = 121 (People clicked from Google to this page 121 times in this month.)
• prev_title = other-google (This data record is for referals from Google.)
• curr_title = !! (This Wikipedia page is about a double exclamation mark.)
• type = other

Create a DataFrame from this CSV

• From the next Spark release - 2.0, CSV as a datasource will be part of Spark's standard release. But, we are using Spark 1.6

// Load the raw dataset stored as a CSV file
val clickstream = sqlContext.
    read.
    format("com.databricks.spark.csv").
    options(Map("header" -> "true", "delimiter" -> "\t", "mode" -> "PERMISSIVE", "inferSchema" -> "true")).
    load("dbfs:///databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed")

clickstream: org.apache.spark.sql.DataFrame = [prev_id: int, curr_id: int ... 4 more fields]

Print the schema

clickstream.printSchema

root
 |-- prev_id: integer (nullable = true)
 |-- curr_id: integer (nullable = true)
 |-- n: integer (nullable = true)
 |-- prev_title: string (nullable = true)
 |-- curr_title: string (nullable = true)
 |-- type: string (nullable = true)

Display some sample data

display(clickstream)

Truncated to 30 rows

Display is a utility provided by Databricks. If you are programming directly in Spark, use the show(numRows: Int) function of DataFrame

clickstream.show(5)

+-------+-------+---+------------------+----------+-----+
|prev_id|curr_id|  n|        prev_title|curr_title| type|
+-------+-------+---+------------------+----------+-----+
|   null|3632887|121|      other-google|        !!|other|
|   null|3632887| 93|   other-wikipedia|        !!|other|
|   null|3632887| 46|       other-empty|        !!|other|
|   null|3632887| 10|       other-other|        !!|other|
|  64486|3632887| 11|!_(disambiguation)|        !!|other|
+-------+-------+---+------------------+----------+-----+
only showing top 5 rows

Reading from disk vs memory

The 1.2 GB Clickstream file is currently on S3, which means each time you scan through it, your Spark cluster has to read the 1.2 GB of data remotely over the network.

Call the count() action to check how many rows are in the DataFrame and to see how long it takes to read the DataFrame from S3.

clickstream.cache().count()

res7: Long = 22509897

• It took about several minutes to read the 1.2 GB file into your Spark cluster. The file has 22.5 million rows/lines.
• Although we have called cache, remember that it is evaluated (cached) only when an action(count) is called

Now call count again to see how much faster it is to read from memory

clickstream.count()

res8: Long = 22509897

• Orders of magnitude faster!
• If you are going to be using the same data source multiple times, it is better to cache it in memory

What are the top 10 articles requested?

To do this we also need to order by the sum of column n, in descending order.

//Type in your answer here...
display(clickstream
  .select(clickstream("curr_title"), clickstream("n"))
  .groupBy("curr_title")
  .sum()
  .orderBy($"sum(n)".desc)
  .limit(10))

Who sent the most traffic to Wikipedia in Feb 2015?

In other words, who were the top referers to Wikipedia?

display(clickstream
  .select(clickstream("prev_title"), clickstream("n"))
  .groupBy("prev_title")
  .sum()
  .orderBy($"sum(n)".desc)
  .limit(10))

As expected, the top referer by a large margin is Google. Next comes refererless traffic (usually clients using HTTPS). The third largest sender of traffic to English Wikipedia are Wikipedia pages that are not in the main namespace (ns = 0) of English Wikipedia. Learn about the Wikipedia namespaces here: https://en.wikipedia.org/wiki/Wikipedia:Project\_namespace

Also, note that Twitter sends 10x more requests to Wikipedia than Facebook.

What were the top 5 trending articles people from Twitter were looking up in Wikipedia?

//Type in your answer here...
display(clickstream
  .select(clickstream("curr_title"), clickstream("prev_title"), clickstream("n"))
  .filter("prev_title = 'other-twitter'")
  .groupBy("curr_title")
  .sum()
  .orderBy($"sum(n)".desc)
  .limit(5))

What percentage of page visits in Wikipedia are from other pages in Wikipedia itself?

val allClicks = clickstream.selectExpr("sum(n)").first.getLong(0)
val referals = clickstream.
                filter(clickstream("prev_id").isNotNull).
                selectExpr("sum(n)").first.getLong(0)
(referals * 100.0) / allClicks

allClicks: Long = 3283067885
referals: Long = 1095462001
res12: Double = 33.36702253416853

Register the DataFrame to perform more complex queries

clickstream.createOrReplaceTempView("clicks")

Which Wikipedia pages have the most referrals to the Donald Trump page?

SELECT *
FROM clicks
WHERE 
  curr_title = 'Donald_Trump' AND
  prev_id IS NOT NULL AND prev_title != 'Main_Page'
ORDER BY n DESC
LIMIT 20

Top referrers to all presidential candidate pages

-- FIXME (broke query, will get back to it later)
SELECT *
FROM clicks
WHERE 
  prev_id IS NOT NULL
ORDER BY n DESC
LIMIT 20

Load a visualization library

This code is copied after doing a live google search (by Michael Armbrust at Spark Summit East February 2016 shared from https://twitter.com/michaelarmbrust/status/699969850475737088). The d3ivan package is an updated version of the original package used by Michael Armbrust as it needed some TLC for Spark 2.2 on newer databricks notebook. These changes were kindly made by Ivan Sadikov from Middle Earth.

Warning: classes defined within packages cannot be redefined without a cluster restart.
Compilation successful.

d3ivan.graphs.help()

Produces a force-directed graph given a collection of edges of the following form:
case class Edge(src: String, dest: String, count: Long)

Usage:
import d3._
graphs.force(
  height = 500,
  width = 500,
  clicks: Dataset[Edge])

d3ivan.graphs.force(
  height = 800,
  width = 1000,
  clicks = sql("""
    SELECT 
      prev_title AS src,
      curr_title AS dest,
      n AS count FROM clicks
    WHERE 
      curr_title IN ('Donald_Trump', 'Bernie_Sanders', 'Hillary_Rodham_Clinton', 'Ted_Cruz') AND
      prev_id IS NOT NULL AND prev_title != 'Main_Page'
    ORDER BY n DESC
    LIMIT 20""").as[d3ivan.Edge])

Convert raw data to parquet

Apache Parquet is a columnar storage format available to any project in the Hadoop ecosystem, regardless of the choice of data processing framework, data model or programming language. It is a more efficient way to store data frames.

• To understand the ideas read Dremel: Interactive Analysis of Web-Scale Datasets, Sergey Melnik, Andrey Gubarev, Jing Jing Long, Geoffrey Romer, Shiva Shivakumar, Matt Tolton and Theo Vassilakis,Proc. of the 36th Int'l Conf on Very Large Data Bases (2010), pp. 330-339, whose Abstract is as follows:
  • Dremel is a scalable, interactive ad-hoc query system for analysis of read-only nested data. By combining multi-level execution trees and columnar data layouts it is capable of running aggregation queries over trillion-row tables in seconds. The system scales to thousands of CPUs and petabytes of data, and has thousands of users at Google. In this paper, we describe the architecture and implementation of Dremel, and explain how it complements MapReduce-based computing. We present a novel columnar storage representation for nested records and discuss experiments on few-thousand node instances of the system.

// Convert the DatFrame to a more efficent format to speed up our analysis
clickstream.
  write.
  mode(SaveMode.Overwrite).
  parquet("/datasets/wiki-clickstream") // warnings are harmless

Load parquet file efficiently and quickly into a DataFrame

Now we can simply load from this parquet file next time instead of creating the RDD from the text file (much slower).

Also using parquet files to store DataFrames allows us to go between languages quickly in a a scalable manner.

val clicks = sqlContext.read.parquet("/datasets/wiki-clickstream")

clicks: org.apache.spark.sql.DataFrame = [prev_id: int, curr_id: int ... 4 more fields]

clicks.printSchema

root
 |-- prev_id: integer (nullable = true)
 |-- curr_id: integer (nullable = true)
 |-- n: integer (nullable = true)
 |-- prev_title: string (nullable = true)
 |-- curr_title: string (nullable = true)
 |-- type: string (nullable = true)

display(clicks)  // let's display this DataFrame

Truncated to 30 rows

DataFrame in python

clicksPy = sqlContext.read.parquet("/datasets/wiki-clickstream")

# in Python you need to put the object int its own line like this to get the type information
clicksPy

Out[2]: DataFrame[prev_id: int, curr_id: int, n: int, prev_title: string, curr_title: string, type: string]

clicksPy.show()

+--------+-------+---+--------------------+--------------------+-----+
| prev_id|curr_id|  n|          prev_title|          curr_title| type|
+--------+-------+---+--------------------+--------------------+-----+
| 7009881| 164003| 21|              Mayall|         John_Mayall| link|
|  476786| 164003| 86|         Mick_Taylor|         John_Mayall| link|
|19735547| 164003| 10|Peter_Green_disco...|         John_Mayall| link|
|  244136| 164003| 10|        Macclesfield|         John_Mayall| link|
|33105755| 164003| 13|       The_Yardbirds|         John_Mayall| link|
| 8910430| 164003| 34|The_Turning_Point...|         John_Mayall| link|
|  329878| 164003| 10|      Steve_Marriott|         John_Mayall| link|
|    null| 164003|652|         other-empty|         John_Mayall|other|
|    null| 147396|134|          other-bing|John_Mayall_&_the...|other|
|17865484| 147396| 13|Timeline_of_heavy...|John_Mayall_&_the...|other|
|15580374| 147396| 94|           Main_Page|John_Mayall_&_the...|other|
|  168254| 147396| 23|    Paul_Butterfield|John_Mayall_&_the...| link|
|  322138| 147396|283|Peter_Green_(musi...|John_Mayall_&_the...| link|
|    null| 147396| 79|         other-other|John_Mayall_&_the...|other|
|12154926| 147396| 13|Marshall_Bluesbre...|John_Mayall_&_the...| link|
|  223910| 147396| 12|         Robben_Ford|John_Mayall_&_the...|other|
|14433637| 147396| 10|Parchman_Farm_(song)|John_Mayall_&_the...| link|
|  476786| 147396|213|         Mick_Taylor|John_Mayall_&_the...| link|
|18952282| 147396| 13|           Ric_Grech|John_Mayall_&_the...|other|
| 4113741| 147396| 50|Rolling_Stone's_5...|John_Mayall_&_the...| link|
+--------+-------+---+--------------------+--------------------+-----+
only showing top 20 rows

Now you can continue from the original python notebook tweeted by Michael.

Recall from the beginning of this notebook that this python databricks notebook was used in the talk by Michael Armbrust at Spark Summit East February 2016 shared from https://twitter.com/michaelarmbrust/status/699969850475737088

(watch now, if you haven't already!)

You Try!

Try to laoad a DataFrame in R from the parquet file just as we did for python. Read the docs in databricks guide first:

• https://docs.databricks.com/spark/latest/sparkr/overview.html

And see the R example in the Programming Guide:

• https://spark.apache.org/docs/latest/sql-programming-guide.html#parquet-files.

library(SparkR)

# just a quick test
df <- createDataFrame(faithful)
head(df)

# Read in the Parquet file created above. Parquet files are self-describing so the schema is preserved.
# The result of loading a parquet file is also a DataFrame.
clicksR <- read.df("/datasets/wiki-clickstream", source = "parquet")
clicksR # in R you need to put the object int its own line like this to get the type information

head(clicksR)

display(clicksR)

Truncated to 30 rows