datakrk-spark.ipynb

{
 "metadata": {
  "name": "",
  "signature": "sha256:b1ab80d415356bd6a6f04d3a4eefe56ea8ebd0abeb6778c0c9ad161bc1fdf350"
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "# BigData\n",
      "## PySpark"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "sc"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 1,
       "text": [
        "<pyspark.context.SparkContext at 0x3728910>"
       ]
      }
     ],
     "prompt_number": 1
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "It runs on top of YARN cluster"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "!yarn node -list"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "l = sc.parallelize(range(1000), 10)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 2
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "**Action** evaluates RDD"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Action\n",
      "l.take(10)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 3,
       "text": [
        "[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]"
       ]
      }
     ],
     "prompt_number": 3
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "**Transformation** creats another RDD, but not evaluated"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Transformation\n",
      "l.map(lambda i: i**2)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 4,
       "text": [
        "PythonRDD[2] at RDD at PythonRDD.scala:43"
       ]
      }
     ],
     "prompt_number": 4
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "l.map(lambda i: i**2).take(10)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 5,
       "text": [
        "[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]"
       ]
      }
     ],
     "prompt_number": 5
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Let's implement word count in PySpark"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "text = sc.textFile(\"s3://YOUR_S3_BUCKET/ulysses.txt\")\n",
      "words = text.flatMap(lambda line: line.split()) \\\n",
      "            .map(lambda word: word.lower())\n",
      "\n",
      "def wordcounts(words_rdd):\n",
      "    return words_rdd.map(lambda word: (word, 1)) \\\n",
      "                .reduceByKey(lambda x,y: x + y) \\\n",
      "                .map(lambda (word, count): (count, word)) \\\n",
      "                .sortByKey(False)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 6
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "wordcounts(words).take(10)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 7,
       "text": [
        "[(14853, u'the'),\n",
        " (8216, u'of'),\n",
        " (7052, u'and'),\n",
        " (6400, u'a'),\n",
        " (4912, u'to'),\n",
        " (4796, u'in'),\n",
        " (3620, u'he'),\n",
        " (3275, u'his'),\n",
        " (2487, u'with'),\n",
        " (2442, u'i')]"
       ]
      }
     ],
     "prompt_number": 7
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Top words are \"stopwords\". Let's filter them out."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "stopwords = set(sc.textFile(\n",
      "                \"s3://YOUR_S3_BUCKET/stopwords.txt\")\n",
      "                .collect())\n",
      "\n",
      "wordcounts(\n",
      "    words.filter(lambda word: word not in stopwords)\n",
      ").take(10)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 8,
       "text": [
        "[(706, u'like'),\n",
        " (701, u'mr'),\n",
        " (573, u'one'),\n",
        " (480, u'said.'),\n",
        " (469, u'old'),\n",
        " (454, u'says'),\n",
        " (442, u'bloom'),\n",
        " (428, u'said'),\n",
        " (345, u'two'),\n",
        " (317, u'see')]"
       ]
      }
     ],
     "prompt_number": 8
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Let's process Adult dataset using Spark"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "adult_data = sc.textFile(\n",
      "                \"s3://YOUR_S3_BUCKET/adult-dataset-data.csv\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 9
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from csv import reader\n",
      "from StringIO import StringIO\n",
      "import re\n",
      "\n",
      "def parse_row(row):\n",
      "    return map(cast_element, row.split(\",\"))\n",
      "    \n",
      "def cast_element(element):\n",
      "    if re.match(r\"^\\d+$\", element):\n",
      "        return int(element)\n",
      "    elif re.match(r\"^\\d+.\\d*$\", element):\n",
      "        return float(element)\n",
      "    else:\n",
      "        return element.strip()\n",
      "        \n",
      "adult_data.map(parse_row).first()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 10,
       "text": [
        "[1,\n",
        " 39.0,\n",
        " u'State-gov',\n",
        " 77516.0,\n",
        " u'Bachelors',\n",
        " 13.0,\n",
        " u'Never-married',\n",
        " u'Adm-clerical',\n",
        " u'Not-in-family',\n",
        " u'White',\n",
        " u'Male',\n",
        " 2174.0,\n",
        " 0.0,\n",
        " 40.0,\n",
        " u'United-States']"
       ]
      }
     ],
     "prompt_number": 10
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "SparkSQL is a powerful way of querying RDD's with SQL-like language."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from pyspark.sql import Row, SQLContext\n",
      "sqlContext = SQLContext(sc)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 11
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "headers = [\"id\", \"age\", \"workclass\", \"final_weight\", \"education\", \"education_num\", \"marital_status\",\n",
      "           \"occupation\", \"relationship\", \"race\", \"sex\", \"capital_gain\", \"capital_loss\", \"hours_per_week\", \"native_country\"]\n",
      "\n",
      "def parse_to_spark_sql_row(row):\n",
      "    return Row(**dict(zip(headers, parse_row(row))))\n",
      "\n",
      "\n",
      "adults_sql = adult_data.map(parse_to_spark_sql_row)\n",
      "adults_sql.first()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 12,
       "text": [
        "Row(age=39.0, capital_gain=2174.0, capital_loss=0.0, education=u'Bachelors', education_num=13.0, final_weight=77516.0, hours_per_week=40.0, id=1, marital_status=u'Never-married', native_country=u'United-States', occupation=u'Adm-clerical', race=u'White', relationship=u'Not-in-family', sex=u'Male', workclass=u'State-gov')"
       ]
      }
     ],
     "prompt_number": 12
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Spark can automatically infer schema of RDD. Once we have schema, we need register it before querying."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "schemaAdults = sqlContext.inferSchema(adults_sql)\n",
      "schemaAdults.registerTempTable(\"adults\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 13
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "sqlContext.sql(\n",
      "\"\"\"SELECT AVG(age) as average_age\n",
      "   FROM adults\n",
      "   WHERE marital_status='Never-married'\"\"\").first()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 14,
       "text": [
        "Row(average_age=28.150987550313584)"
       ]
      }
     ],
     "prompt_number": 14
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "We can also mix Spark SQL with regular Spark API, e.g. group records by country and take counts using SQL, then sort results using regular Spark API."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "countries_counts = sqlContext.sql(\n",
      "\"\"\"SELECT native_country, count(*) as count_\n",
      "   FROM adults where sex='Male'\n",
      "   GROUP BY native_country\"\"\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 15
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "countries_counts.map(lambda row: (row.count_, row.native_country)).sortByKey(False).take(3)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 16,
       "text": [
        "[(19488, u'United-States'), (497, u'Mexico'), (420, u'?')]"
       ]
      }
     ],
     "prompt_number": 16
    }
   ],
   "metadata": {}
  }
 ]
}