jupyter/ETL.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ETL with PySpark SQL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Importing and creating SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import os, sys\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql.functions import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Create a SparkSession. No need to create SparkContext\n",
    "# You automatically get it as part of the SparkSession\n",
    "spark = SparkSession.builder\\\n",
    "                .master(\"local[*]\")\\\n",
    "                .appName(\"ETL\")\\\n",
    "                .config(\"spark.executor.logs.rolling.time.interval\", \"daily\")\\\n",
    "                .getOrCreate()\n",
    "\n",
    "# there are some config you might want to set:\n",
    "# https://spark.apache.org/docs/latest/configuration.html\n",
    "\n",
    "# now we can go to http://localhost:4040 (default port) in order to see Spark's web UI"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setting filesystem and files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#print(os.getcwd()) # show current working directory\n",
    "\n",
    "#datasetDir = \"../../datasets/\" # local files\n",
    "datasetDir = \"hdfs://localhost:19000/\" # hadoop filesystem"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# load all CSV's files from HiggsTwitter dataset (http://snap.stanford.edu/data/higgs-twitter.html)\n",
    "\n",
    "# First, we set the filename\n",
    "file = datasetDir + \"HiggsTwitter/higgs-social_network.edgelist.gz\"\n",
    "# Second, it's recommended to specify the dataframe's schema to avoid spark calculate it\n",
    "schema = StructType([StructField(\"follower\", IntegerType()), StructField(\"followed\", IntegerType())])\n",
    "# Finally, we create the dataframe with previous variables\n",
    "# Also we can specifying the separator with 'sep' (default separator for CSV is ',')\n",
    "socialDF = spark.read.csv(path=file, sep=\" \", schema=schema)\n",
    "\n",
    "file = datasetDir + \"HiggsTwitter/higgs-retweet_network.edgelist.gz\"\n",
    "schema = StructType([StructField(\"tweeter\", IntegerType()), StructField(\"tweeted\", IntegerType()), StructField(\"occur\", IntegerType())])\n",
    "retweetDF = spark.read.csv(path=file, sep=\" \", schema=schema)\n",
    "\n",
    "file = datasetDir + \"HiggsTwitter/higgs-reply_network.edgelist.gz\"\n",
    "schema = StructType([StructField(\"replier\", IntegerType()), StructField(\"replied\", IntegerType()), StructField(\"occur\", IntegerType())])\n",
    "replyDF = spark.read.csv(path=file, sep=\" \", schema=schema)\n",
    "\n",
    "file = datasetDir + \"HiggsTwitter/higgs-mention_network.edgelist.gz\"\n",
    "schema = StructType([StructField(\"mentioner\", IntegerType()), StructField(\"mentioned\", IntegerType()), StructField(\"occur\", IntegerType())])\n",
    "mentionDF = spark.read.csv(path=file, sep=\" \", schema=schema)\n",
    "\n",
    "file = datasetDir + \"HiggsTwitter/higgs-activity_time.txt.gz\"\n",
    "schema = StructType([StructField(\"userA\", IntegerType()), \\\n",
    "                     StructField(\"userB\", IntegerType()), \\\n",
    "                     StructField(\"timestamp\", IntegerType()), \\\n",
    "                    StructField(\"interaction\", StringType())])\n",
    "                    #Interaction can be: RT (retweet), MT (mention) or RE (reply)\n",
    "activityDF = spark.read.csv(path=file, sep=\" \", schema=schema)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Convert CSV's dataframes to Apache Parquet files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "socialDF.write.save(datasetDir + \"HiggsTwitter/higgs-social_network.parquet\")\n",
    "retweetDF.write.save(datasetDir + \"HiggsTwitter/higgs-retweet_network.parquet\")\n",
    "replyDF.write.save(datasetDir + \"HiggsTwitter/higgs-reply_network.parquet\")\n",
    "mentionDF.write.save(datasetDir + \"HiggsTwitter/higgs-mention_network.parquet\")\n",
    "activityDF.write.save(datasetDir + \"HiggsTwitter/higgs-activity_time.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load the parquet files into new dataframes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "socialDFpq = spark.read.load(datasetDir + \"HiggsTwitter/higgs-social_network.parquet\")\n",
    "retweetDFpq = spark.read.load(datasetDir + \"HiggsTwitter/higgs-retweet_network.parquet\")\n",
    "replyDFpq = spark.read.load(datasetDir + \"HiggsTwitter/higgs-reply_network.parquet\")\n",
    "mentionDFpq = spark.read.load(datasetDir + \"HiggsTwitter/higgs-mention_network.parquet\")\n",
    "activityDFpq = spark.read.load(datasetDir + \"HiggsTwitter/higgs-activity_time.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Working with dataframes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- follower: integer (nullable = true)\n",
      " |-- followed: integer (nullable = true)\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "StructType(List(StructField(follower,IntegerType,true),StructField(followed,IntegerType,true)))"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# two ways of showing dataframe's schema\n",
    "socialDFpq.printSchema()\n",
    "socialDFpq.schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------+\n",
      "|follower|followed|\n",
      "+--------+--------+\n",
      "|       1|       2|\n",
      "|       1|       3|\n",
      "|       1|       4|\n",
      "+--------+--------+\n",
      "only showing top 3 rows\n",
      "\n",
      "+-------+-------+-----+\n",
      "|tweeter|tweeted|occur|\n",
      "+-------+-------+-----+\n",
      "| 298960| 105232|    1|\n",
      "|  95688|   3393|    1|\n",
      "| 353237|  62217|    1|\n",
      "+-------+-------+-----+\n",
      "only showing top 3 rows\n",
      "\n",
      "+-------+-------+-----+\n",
      "|replier|replied|occur|\n",
      "+-------+-------+-----+\n",
      "| 161345|   8614|    1|\n",
      "| 428368|  11792|    1|\n",
      "|  77904|  10701|    1|\n",
      "+-------+-------+-----+\n",
      "only showing top 3 rows\n",
      "\n",
      "+---------+---------+-----+\n",
      "|mentioner|mentioned|occur|\n",
      "+---------+---------+-----+\n",
      "|   316609|     5011|    1|\n",
      "|   439696|    12389|    1|\n",
      "|    60059|     6929|    1|\n",
      "+---------+---------+-----+\n",
      "only showing top 3 rows\n",
      "\n",
      "+------+------+----------+-----------+\n",
      "| userA| userB| timestamp|interaction|\n",
      "+------+------+----------+-----------+\n",
      "|223789|213163|1341100972|         MT|\n",
      "|223789|213163|1341100972|         RE|\n",
      "|376989| 50329|1341101181|         RT|\n",
      "+------+------+----------+-----------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# showing some data from dataframes\n",
    "socialDFpq.show(3)\n",
    "retweetDFpq.show(3)\n",
    "replyDFpq.show(3)\n",
    "mentionDFpq.show(3)\n",
    "activityDFpq.show(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark SQL using DataFrames API"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+\n",
      "|followed|followers|\n",
      "+--------+---------+\n",
      "|    1503|    51386|\n",
      "|     206|    48414|\n",
      "|      88|    45221|\n",
      "|     138|    44188|\n",
      "|    1062|    40120|\n",
      "+--------+---------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+---------+--------+\n",
      "|mentioned|mentions|\n",
      "+---------+--------+\n",
      "|       88|   11953|\n",
      "|      677|    3906|\n",
      "|     2417|    2533|\n",
      "|    59195|    1601|\n",
      "|     3998|    1587|\n",
      "+---------+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Users who have most followers\n",
    "socialDFpq.groupBy(\"followed\").agg(count(\"followed\").alias(\"followers\")).orderBy(desc(\"followers\")).show(5)\n",
    "\n",
    "# Users who have most mentions\n",
    "mentionDFpq.groupBy(\"mentioned\").agg(count(\"occur\").alias(\"mentions\")).orderBy(desc(\"mentions\")).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+--------+\n",
      "|followed|followers|mentions|\n",
      "+--------+---------+--------+\n",
      "|    1503|    51386|     150|\n",
      "|     206|    48414|     397|\n",
      "|      88|    45221|   15687|\n",
      "|     138|    44188|     347|\n",
      "|    1062|    40120|      84|\n",
      "+--------+---------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Of the top 5 followed users, how many mentions has each one?\n",
    "\n",
    "# top_f contains \"top 5 users who have most followers\"\n",
    "top_f = socialDFpq.groupBy(\"followed\").agg(count(\"follower\").alias(\"followers\")).orderBy(desc(\"followers\")).limit(5)\n",
    "\n",
    "top_f.join(mentionDFpq, top_f.followed == mentionDFpq.mentioned)\\\n",
    "    .groupBy(top_f.followed, top_f.followers)\\\n",
    "        .agg(sum(mentionDFpq.occur).alias(\"mentions\"))\\\n",
    "    .orderBy(desc(\"followers\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark SQL using SQL language"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# create temporary views so we can use SQL statements\n",
    "socialDFpq.createOrReplaceTempView(\"social\")\n",
    "retweetDFpq.createOrReplaceTempView(\"retweet\")\n",
    "replyDFpq.createOrReplaceTempView(\"reply\")\n",
    "mentionDFpq.createOrReplaceTempView(\"mention\")\n",
    "activityDFpq.createOrReplaceTempView(\"activity\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+\n",
      "|followed|followers|\n",
      "+--------+---------+\n",
      "|    1503|    51386|\n",
      "|     206|    48414|\n",
      "|      88|    45221|\n",
      "|     138|    44188|\n",
      "|    1062|    40120|\n",
      "+--------+---------+\n",
      "only showing top 5 rows\n",
      "\n",
      "+---------+--------+\n",
      "|mentioned|mentions|\n",
      "+---------+--------+\n",
      "|       88|   11953|\n",
      "|      677|    3906|\n",
      "|     2417|    2533|\n",
      "|    59195|    1601|\n",
      "|     3998|    1587|\n",
      "+---------+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Users who have most followers\n",
    "spark.sql(\"select followed, count(follower) as followers from social group by followed order by followers desc\").show(5)\n",
    "\n",
    "# Users who have most mentions\n",
    "spark.sql(\"select mentioned, count(occur) as mentions from mention group by mentioned order by mentions desc\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+--------+\n",
      "|followed|followers|mentions|\n",
      "+--------+---------+--------+\n",
      "|    1503|    51386|     150|\n",
      "|     206|    48414|     397|\n",
      "|      88|    45221|   15687|\n",
      "|     138|    44188|     347|\n",
      "|    1062|    40120|      84|\n",
      "+--------+---------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Of the top 5 followed users, how many mentions has each one?\n",
    "spark.sql(\"\"\"\n",
    "select 5_top_f.followed, 5_top_f.followers, sum(m.occur) as mentions\n",
    "    from \n",
    "        -- subquery that contains top 5 of followed users\n",
    "        (select followed, count(follower) as followers from social group by followed order by followers desc limit 5) 5_top_f, \n",
    "        mention as m\n",
    "    where 5_top_f.followed = m.mentioned\n",
    "    group by 5_top_f.followed, followers\n",
    "    order by followers desc\n",
    "        \"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Performance testing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### GZIP Compressed CSV file vs Parquet file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+\n",
      "|followed|followers|\n",
      "+--------+---------+\n",
      "|    1503|    51386|\n",
      "|     206|    48414|\n",
      "|      88|    45221|\n",
      "|     138|    44188|\n",
      "|    1062|    40120|\n",
      "+--------+---------+\n",
      "only showing top 5 rows\n",
      "\n",
      "Wall time: 20.8 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# GZIP Compressed CSV\n",
    "socialDF.groupBy(\"followed\").agg(count(\"followed\").alias(\"followers\")).orderBy(desc(\"followers\")).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+\n",
      "|followed|followers|\n",
      "+--------+---------+\n",
      "|    1503|    51386|\n",
      "|     206|    48414|\n",
      "|      88|    45221|\n",
      "|     138|    44188|\n",
      "|    1062|    40120|\n",
      "+--------+---------+\n",
      "only showing top 5 rows\n",
      "\n",
      "Wall time: 5.19 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# Parquet file\n",
    "socialDFpq.groupBy(\"followed\").agg(count(\"followed\").alias(\"followers\")).orderBy(desc(\"followers\")).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Cached DF vs not cached DF"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This time we will cache the 2 previous dataframes (socialDF and socialDFpq) and see how faster is."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[follower: int, followed: int]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# cache dataframes\n",
    "socialDF.cache()\n",
    "socialDFpq.cache()\n",
    "\n",
    "# remove from cache\n",
    "#socialDF.unpersist()\n",
    "#socialDFpq.unpersist()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Note: The first time we run cached dataframes can be slower, but the next times they should run faster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+\n",
      "|followed|followers|\n",
      "+--------+---------+\n",
      "|    1503|    51386|\n",
      "|     206|    48414|\n",
      "|      88|    45221|\n",
      "|     138|    44188|\n",
      "|    1062|    40120|\n",
      "+--------+---------+\n",
      "only showing top 5 rows\n",
      "\n",
      "Wall time: 3.82 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# GZIP Compressed CSV (dataframe cached)\n",
    "socialDF.groupBy(\"followed\").agg(count(\"followed\").alias(\"followers\")).orderBy(desc(\"followers\")).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+\n",
      "|followed|followers|\n",
      "+--------+---------+\n",
      "|    1503|    51386|\n",
      "|     206|    48414|\n",
      "|      88|    45221|\n",
      "|     138|    44188|\n",
      "|    1062|    40120|\n",
      "+--------+---------+\n",
      "only showing top 5 rows\n",
      "\n",
      "Wall time: 3.75 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# Parquet file (dataframe cached)\n",
    "socialDFpq.groupBy(\"followed\").agg(count(\"followed\").alias(\"followers\")).orderBy(desc(\"followers\")).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Accessing local files vs local HDFS (Hadoop Distributed File System)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using only 1 cluster should make no real difference accessing local files or local HDFS."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Local file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "datasetDir = \"../../datasets/\" # local files\n",
    "file = datasetDir + \"HiggsTwitter/higgs-social_network.edgelist.gz\"\n",
    "schema = StructType([StructField(\"follower\", IntegerType()), StructField(\"followed\", IntegerType())])\n",
    "\n",
    "socialDFLocal = spark.read.csv(path=file, sep=\" \", schema=schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+\n",
      "|followed|followers|\n",
      "+--------+---------+\n",
      "|    1503|    51386|\n",
      "|     206|    48414|\n",
      "|      88|    45221|\n",
      "|     138|    44188|\n",
      "|    1062|    40120|\n",
      "+--------+---------+\n",
      "only showing top 5 rows\n",
      "\n",
      "Wall time: 16.4 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# Local file with one worker thread\n",
    "socialDFLocal.groupBy(\"followed\").agg(count(\"followed\").alias(\"followers\")).orderBy(desc(\"followers\")).show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Local HDFS file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "datasetDir = \"hdfs://localhost:19000/\" # hadoop filesystem\n",
    "file = datasetDir + \"HiggsTwitter/higgs-social_network.edgelist.gz\"\n",
    "schema = StructType([StructField(\"follower\", IntegerType()), StructField(\"followed\", IntegerType())])\n",
    "\n",
    "socialDFHDFS = spark.read.csv(path=file, sep=\" \", schema=schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+\n",
      "|followed|followers|\n",
      "+--------+---------+\n",
      "|    1503|    51386|\n",
      "|     206|    48414|\n",
      "|      88|    45221|\n",
      "|     138|    44188|\n",
      "|    1062|    40120|\n",
      "+--------+---------+\n",
      "only showing top 5 rows\n",
      "\n",
      "Wall time: 18.2 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# Local file with one worker thread\n",
    "socialDFHDFS.groupBy(\"followed\").agg(count(\"followed\").alias(\"followers\")).orderBy(desc(\"followers\")).show(5)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3.0
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}