ecommercedataset-example.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A demo notebook : Sample ecommerce dataset from AWS"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## This demo illustrates the following \n",
    "\n",
    "* Defining spark external tables on data \n",
    "* Defining the star schema between the data tables \n",
    "* Loading a set of tables into SNAP ( Indexing Process)\n",
    "* Sample queries on SNAP\n",
    "NOTE: This example uses a Jupyter notebook but you can use any SQL client such as Squirrel to connect to SNAP."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### First let us setup the notebook"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "from pyhive import hive\n",
    "from pprint import pprint\n",
    "import pandas as pd\n",
    "import os\n",
    "def sql(q, explain=False) :\n",
    "    # silly hack to handle filesystem prefix for us when creating local tables\n",
    "    if \"{prefix}\" in q:\n",
    "        q = q.replace('{prefix}',cwd)\n",
    "    df=pd.read_sql(q,c)\n",
    "    return df\n",
    "\n",
    "def explain(q):\n",
    "    df = sql(\"explain \" + q)\n",
    "    plan = df['plan'][0]\n",
    "    pprint(plan)\n",
    "    \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Connect to SNAP"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Set up the connection to the SNAP Thrift server\n",
    "c = hive.Connection(host=\"0.0.0.0\",port=10000)\n",
    "pd.read_sql('show tables',c)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Drop all tables and recreate them\n",
    "\n",
    "Note these are external tables - so drop will only delete the metadata not the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "table_names = [\"users\",\"venue\",\"category\",\"ddate\",\"event\",\"listing\",\"sales\"]\n",
    "drop = \"\"\"\n",
    "drop table if exists {table_name}\n",
    "\"\"\"\n",
    "for table_name in table_names:\n",
    "    pd.read_sql(\n",
    "        drop.format(table_name=table_name)\n",
    "    ,c) "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### External table : Users"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "users = \"\"\"\n",
    "create table if not exists users (\n",
    "\tuserid integer ,\n",
    "\tusername string,\n",
    "\tfirstname string,\n",
    "\tlastname string,\n",
    "\tcity string,\n",
    "\tstate string,\n",
    "\temail string,\n",
    "\tphone string,\n",
    "\tlikesports string,\n",
    "\tliketheatre string,\n",
    "\tlikeconcerts string,\n",
    "\tlikejazz string,\n",
    "\tlikeclassical string,\n",
    "\tlikeopera string,\n",
    "\tlikerock string,\n",
    "\tlikevegas string,\n",
    "\tlikebroadway string,\n",
    "\tlikemusicals string)\n",
    "    using csv\n",
    "    options (path \"{prefix}/allusers_pipe.txt\", delimiter \"|\")\n",
    "\"\"\"\n",
    "sql(users)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### External table : Venue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "venue = \"\"\"\n",
    "create table if not exists venue(\n",
    "\tvenueid integer,\n",
    "\tvenuename string,\n",
    "\tvenuecity string,\n",
    "\tvenuestate string,\n",
    "\tvenueseats integer)\n",
    "    using csv\n",
    "    options (path \"{prefix}/venue_pipe.txt\", delimiter \"|\")\n",
    "\"\"\"\n",
    "sql(venue)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### External table : Category"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "category=\"\"\"\n",
    "create table if not exists category(\n",
    "\tcatid integer,\n",
    "\tcatgroup string,\n",
    "\tcatname string,\n",
    "\tcatdesc string)    \n",
    "    using csv\n",
    "    options (path \"{prefix}/category_pipe.txt\", delimiter \"|\")\n",
    "\"\"\"\n",
    "sql(category)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### External table: Date"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "ddate = \"\"\"\n",
    "create table if not exists ddate(\n",
    "\tdateid integer ,\n",
    "\tcaldate date,\n",
    "\tday string,\n",
    "\tweek integer,\n",
    "\tmonth string,\n",
    "\tqtr string,\n",
    "\tyear integer,\n",
    "\tholiday string)    \n",
    "    using csv\n",
    "    options (path \"/data/shared/snap-samples/Redshift/date2008_pipe.txt\", delimiter \"|\")\n",
    "\"\"\"\n",
    "sql(ddate)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### External table : Event"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "event = \"\"\"\n",
    "create table if not exists event(\n",
    "\teventid integer ,\n",
    "\tvenueid integer,\n",
    "\tcatid integer,\n",
    "\tdateid integer ,\n",
    "\teventname string,\n",
    "\tstarttime timestamp)    \n",
    "    using csv\n",
    "    options (path \"{prefix}/allevents_pipe.txt\", delimiter \"|\")\n",
    "\"\"\"\n",
    "sql(event)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### External table: Listing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "listing = \"\"\"\n",
    "create table if not exists listing(\n",
    "\tlistid integer  ,\n",
    "\tsellerid integer ,\n",
    "\teventid integer ,\n",
    "\tdateid integer ,\n",
    "\tnumtickets integer ,\n",
    "\tpriceperticket decimal(8,2),\n",
    "\ttotalprice decimal(8,2),\n",
    "\tlisttime timestamp)    \n",
    "    using csv\n",
    "    options (path \"{prefix}/listings_pipe.txt\", delimiter \"|\")\n",
    "\"\"\"\n",
    "\n",
    "sql(listing)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### External table: Sales"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "sales = \"\"\"\n",
    "create table if not exists  sales(\n",
    "\tsalesid integer ,\n",
    "\tlistid integer,\n",
    "\tsellerid integer,\n",
    "\tbuyerid integer,\n",
    "\teventid integer,\n",
    "\tdateid integer,\n",
    "\tqtysold integer,\n",
    "\tpricepaid decimal(8,2),\n",
    "\tcommission decimal(8,2),\n",
    "\tsaletime timestamp)\n",
    "    using csv\n",
    "    options (\n",
    "        path \"{prefix}/sales_tab.txt\"\n",
    "        ,delimiter \"\\t\"\n",
    "        ,timestampFormat \"d/MM/yyyy HH:mm:ss\"\n",
    "        ,inferSchema \"true\"\n",
    "        )\n",
    "\"\"\"\n",
    "sql(sales)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define the star schema in SNAP\n",
    "The star schema in SNAP establishes the join graph between the external tables we defined above"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "create_star_schema = \"\"\"alter star schema on sales as\n",
    "many_to_one join of sales with listing on sales.listid = listing.listid\n",
    "many_to_one join of sales with event on sales.eventid = event.eventid\n",
    "many_to_one join of sales with ddate on sales.dateid = ddate.dateid\n",
    "many_to_one join of sales with users on sales.buyerid = users.userid\n",
    "many_to_one join of event with category on event.catid = category.catid\n",
    "many_to_one join of event with venue on event.venueid = venue.venueid\n",
    "\"\"\"\n",
    "sql(create_star_schema)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define the SNAP Index "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "salessnap=\"\"\"\n",
    "drop olap index salessnap on sales\n",
    "\"\"\"\n",
    "sql(salessnap)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Index is created with an indexname as \"salessnap\" and we say it is defined on the starschema \"sales\" as defined in the create star schema step."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Dimensions and Metrics\n",
    "Dimensions are columns that you want to filter, select or group by on.\n",
    "Metrics are typically fields of type float/decimal = those you want to aggregate on.\n",
    "timestamp dimensions are a special type of dimension defined on Spark date or timestamp columns\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "salessnap=\"\"\"\n",
    "\n",
    "create olap index salessnap on sales\n",
    "timestamp dimension starttime \n",
    "timestamp dimension listtime \n",
    "timestamp dimension saletime\n",
    "timestamp dimension caldate\n",
    "metric priceperticket aggregator doubleSum is nullable nullvalue \"0.0\" \n",
    "metric totalprice aggregator doubleSum is nullable nullvalue \"0.0\"\n",
    "metric numtickets aggregator longSum is nullable nullvalue \"0\"\n",
    "metric qtysold aggregator longSum is nullable nullvalue \"0\"\n",
    "metric pricepaid aggregator doubleSum is nullable nullvalue \"0.0\"\n",
    "metric commission aggregator doubleSum is nullable nullvalue \"0.0\"\n",
    "dimension holiday is not nullable \n",
    "dimensions \"username,city, state, likesports, liketheatre,likeconcerts, likejazz , \\\n",
    "likeclassical, likeopera, likerock, likevegas, likebroadway, likemusicals, \\\n",
    "venuename, venuecity, venuestate, catgroup, catname, catdesc \\\n",
    ", day, week, month, qtr, year, eventname , \\\n",
    "sales.eventid, sales.buyerid, sales.listid, sales.salesid\" \n",
    "\n",
    "OPTIONS (        \n",
    "    path \"/data/shared/snap-samples/Redshift/snap\",\n",
    "    avgSizePerPartition  \"40mb\",\n",
    "    avgNumRowsPerPartition \"10000\",\n",
    "    preferredSegmentSize \"20mb\",\n",
    "    rowFlushBoundary \"10000\",\n",
    "    defaultNullValue \"0\"\n",
    ")\n",
    "\"\"\"\n",
    "\n",
    "sql(salessnap)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load data into SNAP"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "insert=\"\"\" insert overwrite olap index salessnap of sales \"\"\"\n",
    "\n",
    "sql(insert)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Query 1\n",
    "\n",
    "Note after loading data into SNAP you can query the original Spark external tables sales, ddate with the join keys. The query will get rewritten in SNAP at runtime to use the SNAP index created instead of executing on the source data as joins."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "query1 = \"\"\"\n",
    "SELECT sum(qtysold) \n",
    "FROM   sales as sales, ddate\n",
    "WHERE  sales.dateid = ddate.dateid \n",
    "AND    ddate.caldate = '2008-01-05'\n",
    "\"\"\"\n",
    "sql(query1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "explain(query1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "query2 = \"\"\"\n",
    "SELECT firstname, lastname, total_quantity \n",
    "FROM   (SELECT buyerid, sum(qtysold) total_quantity\n",
    "        FROM  sales\n",
    "        GROUP BY buyerid\n",
    "        ORDER BY total_quantity desc limit 10) Q, users\n",
    "WHERE Q.buyerid = userid\n",
    "ORDER BY Q.total_quantity desc\n",
    "\"\"\"\n",
    "sql(query2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "sq1=\"\"\" \n",
    "\n",
    "SELECT SUM(`sales`.`qtysold`) AS `sum_qtysold_ok`, \n",
    "COUNT(1) AS `x__alias__0` \n",
    "FROM `default`.`sales` `sales` \n",
    "  JOIN `default`.`users` `users` ON (`sales`.`buyerid` = `users`.`userid`) \n",
    "  JOIN `default`.`event` `event` ON (`sales`.`eventid` = `event`.`eventid`) \n",
    "  JOIN `default`.`listing` `listing` ON (`sales`.`listid` = `listing`.`listid`) \n",
    "  JOIN `default`.`ddate` `ddate` ON (`sales`.`dateid` = `ddate`.`dateid`) \n",
    "  JOIN `default`.`category` `category` ON (`event`.`catid` = `category`.`catid`) \n",
    "  JOIN `default`.`venue` `venue` ON (`event`.`venueid` = `venue`.`venueid`) \n",
    "  HAVING (COUNT(1) > 0)\n",
    "\n",
    "\"\"\"\n",
    "sql(sq1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "t10=\"\"\"\n",
    "select  sales.eventid, sum(sales.pricepaid) \n",
    "from sales, event\n",
    "where sales.eventid = event.eventid\n",
    "and sales.pricepaid > 30\n",
    "group by sales.eventid\n",
    "order by 2 desc\n",
    "limit 10\n",
    "\"\"\"\n",
    "\n",
    "\n",
    "\n",
    "sql(t10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "\n",
    "\n",
    "### Create a derived view representing a segment of users"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "sportsandjazz = \"\"\"\n",
    "create or replace view sportsandjazz as \n",
    "select * from salessnap where likesports=\"TRUE\" and likejazz=\"TRUE\"\n",
    "\"\"\"\n",
    "sql(sportsandjazz)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "\n",
    "q=\"\"\"\n",
    "select count(*) from sportsandjazz\n",
    "\"\"\"\n",
    "sql(q)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Compare metrics for all users vs a segment of users"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "q=\"\"\"\n",
    "with allusers AS ( \n",
    "select caldate adate, city,sum(qtysold) q, sum(pricepaid) p \n",
    "from salessnap group by caldate,city)\n",
    ",\n",
    "someusers AS (\n",
    "select caldate sdate,city, sum(qtysold) a, sum(pricepaid) b \n",
    "from salessnap where likeconcerts='TRUE' AND likejazz='TRUE' group by caldate,city)\n",
    "\n",
    "select adate,allusers.city, a, b, round(a/q,2)*100 qratio , round(b/p,2)*100 pratio\n",
    "from allusers, someusers where adate=sdate order by pratio desc limit 5000\n",
    "\"\"\"\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "df=sql(q)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Combine results from SNAP with Pandas for descriptive analysis"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "df[['a','b','qratio','pratio']].describe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "inputHidden": false,
    "outputHidden": false
   },
   "outputs": [],
   "source": [
    "sql(\"\"\" export model on sales to '/tmp/a' \"\"\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernel_info": {
   "name": "python2"
  },
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.13"
  },
  "nteract": {
   "version": "0.4.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}