SparkR Demo.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# SparkR setup up instructions and examples.\n",
    "    ### Questions on this doc: people"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Stop Spark: On top for easy access and because it is really important\n",
    "sparkR.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Setup Procedures"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# finds SparkR in Hadoop\n",
    "library(SparkR, lib.loc = c(file.path(Sys.getenv(\"SPARK_HOME\"), \"R\", \"lib\")))\n",
    "\n",
    "# Create a SparkR session...connecting to Spark on hadoop from the edge node\n",
    "sparkR.session(appName = \"Sparkr Demo\" , \n",
    "               master = \"yarn\", \n",
    "               sparkConfig = list(spark.driver.memory = \"6g\", spark.executor.memory = \"26g\" , spark.driver.cores = \"50\"))\n",
    "\n",
    "# finds Hive Metadata warehouse on hadoop\n",
    "sparkR.session(enableHiveSupport = TRUE)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Call packages or install as needed...these get installed in Miniconda R folder\n",
    "#install.packages('magrittr')  # install.packages\n",
    "#library('dplyr')  # Loads library\n",
    "#library('magrittr')\n",
    "\n",
    "search()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Point to Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "demo_data <- sql(\"\n",
    "select *\n",
    "from prd_sed_fnd.cal_date_dim_v\n",
    "\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# this tells you if you have a SparkDataFrame and the what types the variables are. Don't really need to do this much\n",
    "# str(demo_data)\n",
    "# just the column names. You can copy and paste into code, wich is nice sometimes to remove errors.\n",
    "# colnames(demo_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Take the first NUM rows of a SparkDataFrame and return the results as a R data.frame\n",
    "take(demo_data,5)\n",
    "class(take(demo_data,5))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "# Persist data if you will be using the SparkDataFrame over and over.  Keeps the data in memory on the hadoop.\n",
    "persist(demo_data, \"MEMORY_ONLY\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "source": [
    "### Spark SQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Creating View in Hive to SQL the SparkDataFrames in Spark\n",
    "#createOrReplaceTempView(df, \"df\")\n",
    "createOrReplaceTempView(demo_data, \"demo_data\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# You can now use \"demo_data\" to call the SparkDataFrame in an SQL statement"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "yr_2016 <- SparkR::sql(\"\n",
    "select * \n",
    "from demo_data\n",
    "where acct_yr_i = 2016\n",
    "order by greg_d\n",
    "\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# yr_2016 is also a SparkDataFrame, which means it is distributed\n",
    "class(yr_2016)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "source": [
    "### Bringing Data out of hadoop down to the edge node: collect()/take()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# WITH BIG DATA COMES BIG RESPONSIBILITY !!!!!\n",
    "# Only collect() if you know your data set's size, and the size will not crash the edge node!!!!\n",
    "# Collect() brings all your data to the local edge node and converts it into an R data frame...no longer SparkDataFrame (i.e. distributed)\n",
    "SparkR::collect(yr_2016)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "class(SparkR::collect(yr_2016))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "### Creating a Hive table from a SparkDataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Creating View to SQL the SparkDataFrames in Spark\n",
    "createOrReplaceTempView(yr_2016, \"yr_2016_UP\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Create Table with common SQL syntax\n",
    "sql( 'CREATE TABLE IF NOT EXISTS MI_TEST.YR_2016 AS SELECT * FROM yr_2016_UP')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Drop table in Hive\n",
    "sql( 'DROP TABLE MI_TEST.YR_2016')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "source": [
    "### Join SparkDataFrames"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Inner Join\n",
    "# join(df1, df2, df1$col1 == df2$col2) # Performs an inner join based on expression\n",
    "yr <- sql('SELECT GREG_D , ACCT_YR_I FROM demo_data WHERE ACCT_YR_I = 2016')\n",
    "mth <- sql('SELECT GREG_D , ACCT_MO_I FROM demo_data WHERE ACCT_YR_I = 2016')\n",
    "\n",
    "createOrReplaceTempView(yr, \"yr\")\n",
    "createOrReplaceTempView(mth, \"mth\")\n",
    "\n",
    "classic_join <- sql('SELECT * \n",
    "                     FROM yr as A\n",
    "                     INNER JOIN mth AS B\n",
    "                     ON A.GREG_D = B.GREG_D\n",
    "')\n",
    "\n",
    "SparkR_join <- join(yr, mth, yr$GREG_D == mth$GREG_D) # creates inner join, use joinExpr = 'left' for left join\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "identical(collect(classic_join), collect(SparkR_join))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Remove views and data in memory. demo_data\n",
    "dropTempView('demo_data')\n",
    "dropTempView('yr_2016_UP')\n",
    "dropTempView('yr')\n",
    "dropTempView('mth')\n",
    "unpersist(demo_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sessionInfo()"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "R",
   "language": "R",
   "name": "ir"
  },
  "language_info": {
   "codemirror_mode": "r",
   "file_extension": ".r",
   "mimetype": "text/x-r-source",
   "name": "R",
   "pygments_lexer": "r",
   "version": "3.3.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}