spark-example-sparksql.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Jayson Francis - Exploring sparksql and spark dataframes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Creates a new SQLContext from the SparkContext sc\n",
    "sqlsc = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Creates a Spark DataFrame for the Postgres table, gameclicks\n",
    "df = sqlsc.read.format(\"jdbc\") \\\n",
    "  .option(\"url\", \"jdbc:postgresql://localhost/cloudera?user=cloudera\") \\\n",
    "  .option(\"dbtable\", \"gameclicks\") \\\n",
    "  .load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- timestamp: timestamp (nullable = false)\n",
      " |-- clickid: integer (nullable = false)\n",
      " |-- userid: integer (nullable = false)\n",
      " |-- usersessionid: integer (nullable = false)\n",
      " |-- ishit: integer (nullable = false)\n",
      " |-- teamid: integer (nullable = false)\n",
      " |-- teamlevel: integer (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Display the schema \n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "755806"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Amount of entries in the dataframe\n",
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-------+------+-------------+-----+------+---------+\n",
      "|           timestamp|clickid|userid|usersessionid|ishit|teamid|teamlevel|\n",
      "+--------------------+-------+------+-------------+-----+------+---------+\n",
      "|2016-05-26 15:06:...|    105|  1038|         5916|    0|    25|        1|\n",
      "|2016-05-26 15:07:...|    154|  1099|         5898|    0|    44|        1|\n",
      "|2016-05-26 15:07:...|    229|   899|         5757|    0|    71|        1|\n",
      "|2016-05-26 15:07:...|    322|  2197|         5854|    0|    99|        1|\n",
      "|2016-05-26 15:07:...|     22|  1362|         5739|    0|    13|        1|\n",
      "+--------------------+-------+------+-------------+-----+------+---------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Preview the first 5 rows\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+---------+\n",
      "|userid|teamlevel|\n",
      "+------+---------+\n",
      "|  1038|        1|\n",
      "|  1099|        1|\n",
      "|   899|        1|\n",
      "|  2197|        1|\n",
      "|  1362|        1|\n",
      "+------+---------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Filter columns in Data Frame\n",
    "df.select(\"userid\", \"teamlevel\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+---------+\n",
      "|userid|teamlevel|\n",
      "+------+---------+\n",
      "|  1513|        2|\n",
      "|   868|        2|\n",
      "|  1453|        2|\n",
      "|  1282|        2|\n",
      "|  1473|        2|\n",
      "+------+---------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Filter rows based on criteria.\n",
    "df.filter(df[\"teamlevel\"] > 1).select(\"userid\", \"teamlevel\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------+\n",
      "|ishit| count|\n",
      "+-----+------+\n",
      "|    0|672423|\n",
      "|    1| 83383|\n",
      "+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Group by a column and count\n",
    "df.groupBy(\"ishit\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+----------+\n",
      "|        avg(ishit)|sum(ishit)|\n",
      "+------------------+----------+\n",
      "|0.1103232840173272|     83383|\n",
      "+------------------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Calculate the average and sum\n",
    "from pyspark.sql.functions import *\n",
    "df.select(mean('ishit'), sum('ishit')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Join two dataframes\n",
    "df2 = sqlsc.read.format(\"jdbc\")\\\n",
    "    .option(\"url\", \"jdbc:postgresql://localhost/cloudera?user=cloudera\")\\\n",
    "    .option(\"dbtable\", \"adclicks\")\\\n",
    "    .load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- timestamp: timestamp (nullable = false)\n",
      " |-- txid: integer (nullable = false)\n",
      " |-- usersessionid: integer (nullable = false)\n",
      " |-- teamid: integer (nullable = false)\n",
      " |-- userid: integer (nullable = false)\n",
      " |-- adid: integer (nullable = false)\n",
      " |-- adcategory: string (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df2.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# We can see that df and df2 have a column called 'userid'\n",
    "merge = df.join(df2, 'userid')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- userid: integer (nullable = false)\n",
      " |-- timestamp: timestamp (nullable = false)\n",
      " |-- clickid: integer (nullable = false)\n",
      " |-- usersessionid: integer (nullable = false)\n",
      " |-- ishit: integer (nullable = false)\n",
      " |-- teamid: integer (nullable = false)\n",
      " |-- teamlevel: integer (nullable = false)\n",
      " |-- timestamp: timestamp (nullable = false)\n",
      " |-- txid: integer (nullable = false)\n",
      " |-- usersessionid: integer (nullable = false)\n",
      " |-- teamid: integer (nullable = false)\n",
      " |-- adid: integer (nullable = false)\n",
      " |-- adcategory: string (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "merge.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------------------+-------+-------------+-----+------+---------+--------------------+-----+-------------+------+----+----------+\n",
      "|userid|           timestamp|clickid|usersessionid|ishit|teamid|teamlevel|           timestamp| txid|usersessionid|teamid|adid|adcategory|\n",
      "+------+--------------------+-------+-------------+-----+------+---------+--------------------+-----+-------------+------+----+----------+\n",
      "|   231|2016-06-08 00:45:...| 376796|        23626|    0|   142|        4|2016-06-08 01:40:...|23669|        23626|   142|  27|     games|\n",
      "|   231|2016-06-08 00:45:...| 376796|        23626|    0|   142|        4|2016-06-08 09:24:...|24122|        23626|   142|   4|     games|\n",
      "|   231|2016-06-08 00:45:...| 376796|        23626|    0|   142|        4|2016-06-08 17:21:...|24659|        23626|   142|  22| computers|\n",
      "|   231|2016-06-08 00:45:...| 376796|        23626|    0|   142|        4|2016-06-08 23:34:...|25076|        23626|   142|  21|    movies|\n",
      "|   231|2016-06-08 00:45:...| 376796|        23626|    0|   142|        4|2016-06-09 16:32:...|26220|        23626|   142|  16|  clothing|\n",
      "+------+--------------------+-------+-------------+-----+------+---------+--------------------+-----+-------------+------+----+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "merge.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}