Spark_Test.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spark Test\n",
    "*Danny Luo*\n",
    "\n",
    "The following tutorial tests the basic capabilities of spark and S3 I/O. This notebook is written for Spark 2.0.2, it will not work for Spark 1.x since it uses integrated spark-csv in the S3 I/O steps. Parts of this notebook are modified from this [tutorial](http://blog.insightdatalabs.com/jupyter-on-apache-spark-step-by-step/)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.context.SparkContext at 0x7f66d665ab10>"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Checking if Spark Context is running\n",
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.sql.context.SQLContext at 0x7f66b44ac550>"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Checking if SQL Context is running\n",
    "sqlCtx"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "20"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Parallelizing a simple array with 20 partitions  over your workers\n",
    "rdd = sc.parallelize(range(1000), 20)  \n",
    "rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1000"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Caching an RDD will let it persist in the workers'  memory, only do this with data you expect to use often\n",
    "#You should now be able to see the rdd \"my_rdd\" under the storage tab on the 4040 Spark Admin UI\n",
    "rdd.setName(\"my_rdd\").cache()\n",
    "#Performing a test Action, cache is lazily evaluated so it will not actually cache until you perform an action\n",
    "rdd.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## S3\n",
    "\n",
    "Now we will try to import our practice dataset `iris_data.csv` on our S3 Bucket into Spark as an RDD. Modify the S3 path to your file as necessary. The syntax is `s3n://yourbucketname//path/to/file`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'sepal_length,sepal_width,petal_length,petal_width,species',\n",
       " u'5.1,3.5,1.4,0.2,setosa',\n",
       " u'4.9,3,1.4,0.2,setosa',\n",
       " u'4.7,3.2,1.3,0.2,setosa',\n",
       " u'4.6,3.1,1.5,0.2,setosa']"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#First we will load it in as a text file\n",
    "iris_raw_RDD = sc.textFile('s3n://BucketName/iris_data.csv')\n",
    "iris_raw_RDD.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "That was pretty cool, but let's see if we can read it in as an csv. You can try, as an exercise in Spark transformations and actions, to turn the above raw textfile in a dataset, but we will simply use a handy exisiting library."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-----------+------------+-----------+-------+\n",
      "|sepal_length|sepal_width|petal_length|petal_width|species|\n",
      "+------------+-----------+------------+-----------+-------+\n",
      "|         5.1|        3.5|         1.4|        0.2| setosa|\n",
      "|         4.9|          3|         1.4|        0.2| setosa|\n",
      "|         4.7|        3.2|         1.3|        0.2| setosa|\n",
      "|         4.6|        3.1|         1.5|        0.2| setosa|\n",
      "|           5|        3.6|         1.4|        0.2| setosa|\n",
      "|         5.4|        3.9|         1.7|        0.4| setosa|\n",
      "|         4.6|        3.4|         1.4|        0.3| setosa|\n",
      "|           5|        3.4|         1.5|        0.2| setosa|\n",
      "|         4.4|        2.9|         1.4|        0.2| setosa|\n",
      "|         4.9|        3.1|         1.5|        0.1| setosa|\n",
      "|         5.4|        3.7|         1.5|        0.2| setosa|\n",
      "|         4.8|        3.4|         1.6|        0.2| setosa|\n",
      "|         4.8|          3|         1.4|        0.1| setosa|\n",
      "|         4.3|          3|         1.1|        0.1| setosa|\n",
      "|         5.8|          4|         1.2|        0.2| setosa|\n",
      "|         5.7|        4.4|         1.5|        0.4| setosa|\n",
      "|         5.4|        3.9|         1.3|        0.4| setosa|\n",
      "|         5.1|        3.5|         1.4|        0.3| setosa|\n",
      "|         5.7|        3.8|         1.7|        0.3| setosa|\n",
      "|         5.1|        3.8|         1.5|        0.3| setosa|\n",
      "+------------+-----------+------------+-----------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Note this is Spark 2.0+ command with spark-csv built in. \n",
    "iris_df = spark.read.csv(\"s3n://BucketName/iris_data.csv\", header=True)\n",
    "\n",
    "iris_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{u'count': u'150',\n",
       " u'max': u'7.9',\n",
       " u'mean': u'5.843333333333335',\n",
       " u'min': u'4.3',\n",
       " u'stddev': u'0.8280661279778637'}"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Generating some statistics for sepal length.\n",
    "iris_stats_df = iris_df.describe('sepal_length').rdd.collectAsMap()\n",
    "iris_stats_df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "Now we will try uploading the `iris_data` back on S3."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "#Saving locally does not seem to work in the Jupyter environment but it does work in PySpark shell. \n",
    "#In Jupyter it creates an empty repository (with _SUCCESS indicator) if you try to save locally.\n",
    "\n",
    "#However, this will work if you save it on s3\n",
    "iris_df.write.csv('s3n://BucketName/iris_data_2')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "Go on your S3 console, or in your AWS CLI, to check to see if the file has been uploaded properly. It will be saved in partitions. In this case, only one partition is created."
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python [conda root]",
   "language": "python",
   "name": "conda-root-py"
  },
  "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}