Spark DataFrames & SQL - Basics.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spark DataFrames & SQL - Basics\n",
    "\n",
    "Xiaodong DENG \n",
    "\n",
    "[xd.deng.r@gmail.com](mailto:xd.deng.r@gmail.com)\n",
    "\n",
    "https://github.com/XD-DENG"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import division\n",
    "\n",
    "import pyspark\n",
    "sc = pyspark.SparkContext()\n",
    "sqlContext = pyspark.SQLContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Build Spark DataFrames"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.1 Build Spark DataFrames from Python Lists"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+---+\n",
      "|  A|  B|\n",
      "+---+---+\n",
      "|  1|  2|\n",
      "|  3|  4|\n",
      "+---+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Build a dataframe from Python Lists.\n",
    "\n",
    "DT1 = sqlContext.createDataFrame(data=[(1,2), (3,4)], schema=(\"A\", \"B\"))\n",
    "\n",
    "DT1.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.2 Build Spark DataFrames from RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['\"date\",\"time\",\"size\",\"r_version\",\"r_arch\",\"r_os\",\"package\",\"version\",\"country\",\"ip_id\"',\n",
       " '\"2015-12-12\",\"13:42:10\",257886,\"3.2.2\",\"i386\",\"mingw32\",\"HistData\",\"0.7-6\",\"CZ\",1',\n",
       " '\"2015-12-12\",\"13:24:37\",1236751,\"3.2.2\",\"x86_64\",\"mingw32\",\"RJSONIO\",\"1.3-0\",\"DE\",2',\n",
       " '\"2015-12-12\",\"13:42:35\",2077876,\"3.2.2\",\"i386\",\"mingw32\",\"UsingR\",\"2.0-5\",\"CZ\",1']"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.textFile(\"sample_data/2015-12-12.csv\",\n",
    "            use_unicode=True).\\\n",
    "take(4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "dat = sc.textFile(\"sample_data/2015-12-12.csv\", use_unicode=True) \\\n",
    "                    .map(lambda x:x.replace('\"', \"\")) \\\n",
    "                    .map(lambda x:x.split(\",\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[['date',\n",
       "  'time',\n",
       "  'size',\n",
       "  'r_version',\n",
       "  'r_arch',\n",
       "  'r_os',\n",
       "  'package',\n",
       "  'version',\n",
       "  'country',\n",
       "  'ip_id'],\n",
       " ['2015-12-12',\n",
       "  '13:42:10',\n",
       "  '257886',\n",
       "  '3.2.2',\n",
       "  'i386',\n",
       "  'mingw32',\n",
       "  'HistData',\n",
       "  '0.7-6',\n",
       "  'CZ',\n",
       "  '1']]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dat.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Note**: there can not be \".\" in the column names (header)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[date: string, time: string, size: string, r_version: string, r_arch: string, r_os: string, package: string, version: string, country: string, ip_id: string]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "DT2 = sqlContext.createDataFrame(data = dat.filter(lambda x:x[0]!='date'),\n",
    "                                 schema=dat.filter(lambda x:x[0]=='date').\\\n",
    "                                 collect()[0])\n",
    "\n",
    "DT2.persist()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+-------+---------+------+-------+---------+-------+-------+-----+\n",
      "|      date|    time|   size|r_version|r_arch|   r_os|  package|version|country|ip_id|\n",
      "+----------+--------+-------+---------+------+-------+---------+-------+-------+-----+\n",
      "|2015-12-12|13:42:10| 257886|    3.2.2|  i386|mingw32| HistData|  0.7-6|     CZ|    1|\n",
      "|2015-12-12|13:24:37|1236751|    3.2.2|x86_64|mingw32|  RJSONIO|  1.3-0|     DE|    2|\n",
      "|2015-12-12|13:42:35|2077876|    3.2.2|  i386|mingw32|   UsingR|  2.0-5|     CZ|    1|\n",
      "|2015-12-12|13:42:01| 266724|    3.2.2|  i386|mingw32|gridExtra|  2.0.0|     CZ|    1|\n",
      "|2015-12-12|13:00:21|3687766|       NA|    NA|     NA|     lme4| 1.1-10|     DE|    3|\n",
      "|2015-12-12|13:08:56|  57429|       NA|    NA|     NA| testthat| 0.11.0|     DE|    3|\n",
      "|2015-12-12|13:08:09| 216068|    3.2.2|x86_64|mingw32|  mvtnorm|  1.0-3|     DE|    4|\n",
      "|2015-12-12|13:25:00|3595497|    3.2.2|x86_64|mingw32|     maps|  3.0.1|     DE|    2|\n",
      "|2015-12-12|13:25:05|1579597|    3.2.2|x86_64|mingw32|       sp|  1.2-1|     DE|    2|\n",
      "|2015-12-12|13:25:21| 892152|    3.2.3|x86_64|mingw32|geosphere|  1.4-3|     DE|    2|\n",
      "+----------+--------+-------+---------+------+-------+---------+-------+-------+-----+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "DT2.show(n = 10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['date',\n",
       " 'time',\n",
       " 'size',\n",
       " 'r_version',\n",
       " 'r_arch',\n",
       " 'r_os',\n",
       " 'package',\n",
       " 'version',\n",
       " 'country',\n",
       " 'ip_id']"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "DT2.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('date', 'string'),\n",
       " ('time', 'string'),\n",
       " ('size', 'string'),\n",
       " ('r_version', 'string'),\n",
       " ('r_arch', 'string'),\n",
       " ('r_os', 'string'),\n",
       " ('package', 'string'),\n",
       " ('version', 'string'),\n",
       " ('country', 'string'),\n",
       " ('ip_id', 'string')]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "DT2.dtypes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Change DataFrames Properties\n",
    "\n",
    "### 2.1 Change Column Type\n",
    "\n",
    "Availabel types include\n",
    "- BinaryType\n",
    "- BooleanType\n",
    "- ByteType\n",
    "- DoubleType\n",
    "- DateType\n",
    "- FloatType\n",
    "- IntegerType\n",
    "- etc."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import IntegerType, DateType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "# .withColumn return a DataFrame by adding a column or replacing the existing column that has the same name.\n",
    "DT3 = DT2.withColumn(\"size\", DT2[\"size\"].cast(IntegerType()))\n",
    "DT3 = DT3.withColumn(\"date\", DT3[\"date\"].cast(DateType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('date', 'date'),\n",
       " ('time', 'string'),\n",
       " ('size', 'int'),\n",
       " ('r_version', 'string'),\n",
       " ('r_arch', 'string'),\n",
       " ('r_os', 'string'),\n",
       " ('package', 'string'),\n",
       " ('version', 'string'),\n",
       " ('country', 'string'),\n",
       " ('ip_id', 'string')]"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "DT3.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+-------+---------+------+-------+---------+-------+-------+-----+\n",
      "|      date|    time|   size|r_version|r_arch|   r_os|  package|version|country|ip_id|\n",
      "+----------+--------+-------+---------+------+-------+---------+-------+-------+-----+\n",
      "|2015-12-12|13:42:10| 257886|    3.2.2|  i386|mingw32| HistData|  0.7-6|     CZ|    1|\n",
      "|2015-12-12|13:24:37|1236751|    3.2.2|x86_64|mingw32|  RJSONIO|  1.3-0|     DE|    2|\n",
      "|2015-12-12|13:42:35|2077876|    3.2.2|  i386|mingw32|   UsingR|  2.0-5|     CZ|    1|\n",
      "|2015-12-12|13:42:01| 266724|    3.2.2|  i386|mingw32|gridExtra|  2.0.0|     CZ|    1|\n",
      "|2015-12-12|13:00:21|3687766|       NA|    NA|     NA|     lme4| 1.1-10|     DE|    3|\n",
      "+----------+--------+-------+---------+------+-------+---------+-------+-------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "DT3.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.2 Change Column Name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "# .withColumnRenamed(existing, new) returns a new DataFrame by renaming an existing column.\n",
    "DT4 = DT2.withColumnRenamed(\"size\", \"size_new\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+--------+---------+------+-------+---------+-------+-------+-----+\n",
      "|      date|    time|size_new|r_version|r_arch|   r_os|  package|version|country|ip_id|\n",
      "+----------+--------+--------+---------+------+-------+---------+-------+-------+-----+\n",
      "|2015-12-12|13:42:10|  257886|    3.2.2|  i386|mingw32| HistData|  0.7-6|     CZ|    1|\n",
      "|2015-12-12|13:24:37| 1236751|    3.2.2|x86_64|mingw32|  RJSONIO|  1.3-0|     DE|    2|\n",
      "|2015-12-12|13:42:35| 2077876|    3.2.2|  i386|mingw32|   UsingR|  2.0-5|     CZ|    1|\n",
      "|2015-12-12|13:42:01|  266724|    3.2.2|  i386|mingw32|gridExtra|  2.0.0|     CZ|    1|\n",
      "|2015-12-12|13:00:21| 3687766|       NA|    NA|     NA|     lme4| 1.1-10|     DE|    3|\n",
      "+----------+--------+--------+---------+------+-------+---------+-------+-------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "DT4.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.3 Order DataFrame by Specified Column(s)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+----+---------+------+----+------------+-------+-------+-----+\n",
      "|      date|    time|size|r_version|r_arch|r_os|     package|version|country|ip_id|\n",
      "+----------+--------+----+---------+------+----+------------+-------+-------+-----+\n",
      "|2015-12-12|20:12:20| 504|       NA|    NA|  NA|  KernSmooth| 2.23-4|     CN|13246|\n",
      "|2015-12-12|19:06:56| 504|       NA|    NA|  NA| httpRequest|  0.0.5|     CN| 1133|\n",
      "|2015-12-12|20:33:36| 504|       NA|    NA|  NA|influence.ME|    0.7|     CN| 5337|\n",
      "|2015-12-12|20:34:41| 504|       NA|    NA|  NA|      merror|    1.0|     CN| 5337|\n",
      "|2015-12-12|20:33:02| 504|       NA|    NA|  NA|     granova|    1.4|     CN| 5331|\n",
      "|2015-12-12|20:36:24| 504|       NA|    NA|  NA|       pheno|    1.5|     CN| 4943|\n",
      "|2015-12-12|20:20:28| 504|       NA|    NA|  NA|   orloca.es|    3.2|     CN| 2365|\n",
      "|2015-12-12|20:20:58| 504|       NA|    NA|  NA| poistweedie|    1.0|     CN|   74|\n",
      "|2015-12-12|19:09:58| 504|       NA|    NA|  NA|     polycor|  0.7-0|     CN| 1153|\n",
      "|2015-12-12|20:35:15| 504|       NA|    NA|  NA|      muStat|  1.6.0|     CN| 4943|\n",
      "+----------+--------+----+---------+------+----+------------+-------+-------+-----+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "DT3.sort(DT3.size.asc()).show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+--------+---------+------+------------+--------------------+-------+-------+-----+\n",
      "|      date|    time|    size|r_version|r_arch|        r_os|             package|version|country|ip_id|\n",
      "+----------+--------+--------+---------+------+------------+--------------------+-------+-------+-----+\n",
      "|2015-12-12|02:27:31|68736865|    3.2.3|x86_64|     mingw32|           TCGA2STAT|    1.2|     US| 2700|\n",
      "|2015-12-12|01:31:52|68736865|    3.1.3|x86_64|darwin10.8.0|           TCGA2STAT|    1.2|     US| 2700|\n",
      "|2015-12-12|02:28:49|68736865|    3.3.0|x86_64|   linux-gnu|           TCGA2STAT|    1.2|     US| 2700|\n",
      "|2015-12-12|02:28:30|68736865|    3.2.3|x86_64|     mingw32|           TCGA2STAT|    1.2|     US| 2700|\n",
      "|2015-12-12|21:23:23|68736862|    3.2.3|x86_64|darwin13.4.0|           TCGA2STAT|    1.2|     US| 2700|\n",
      "|2015-12-12|02:19:32|68736862|    3.2.0|  i386|     mingw32|           TCGA2STAT|    1.2|     US| 2700|\n",
      "|2015-12-12|01:28:03|68736856|    3.2.3|x86_64|     mingw32|           TCGA2STAT|    1.2|     US| 2700|\n",
      "|2015-12-12|13:17:41|68736856|    3.2.3|x86_64|   linux-gnu|           TCGA2STAT|    1.2|     GB|  548|\n",
      "|2015-12-12|17:06:20|68736856|       NA|    NA|          NA|           TCGA2STAT|    1.2|     US| 3084|\n",
      "|2015-12-12|01:41:08|62559786|       NA|    NA|          NA|ChemometricsWithR...|  0.1.3|     US| 7666|\n",
      "+----------+--------+--------+---------+------+------------+--------------------+-------+-------+-----+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "DT3.sort(DT3.size.desc()).show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Filtering, and Aggregation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.11161009458040756"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "DT3.filter(DT3['size'] <1000).count() / DT3.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.009273193054466087"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "DT3.filter(DT3['package'] == \"ggplot2\").count() / DT3.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-----+\n",
      "|     package|count|\n",
      "+------------+-----+\n",
      "|        Rcpp| 4783|\n",
      "|     ggplot2| 3913|\n",
      "|     stringi| 3748|\n",
      "|     stringr| 3449|\n",
      "|        plyr| 3436|\n",
      "|    magrittr| 3265|\n",
      "|      digest| 3223|\n",
      "|    reshape2| 3205|\n",
      "|RColorBrewer| 3046|\n",
      "|      scales| 3007|\n",
      "+------------+-----+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "DT3.groupBy(\"package\").count().sort(\"count\", ascending = False).show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "package_count = DT3.groupBy(\"package\").count().sort(\"count\", ascending = False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-----+\n",
      "|     package|count|\n",
      "+------------+-----+\n",
      "|        Rcpp| 4783|\n",
      "|     ggplot2| 3913|\n",
      "|     stringi| 3748|\n",
      "|     stringr| 3449|\n",
      "|        plyr| 3436|\n",
      "|    magrittr| 3265|\n",
      "|      digest| 3223|\n",
      "|    reshape2| 3205|\n",
      "|RColorBrewer| 3046|\n",
      "|      scales| 3007|\n",
      "+------------+-----+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "package_count.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Transform A DataFrame Column (using UDF)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-----+------+\n",
      "|     package|count|  perc|\n",
      "+------------+-----+------+\n",
      "|        Rcpp| 4783|1.133%|\n",
      "|     ggplot2| 3913|0.927%|\n",
      "|     stringi| 3748|0.888%|\n",
      "|     stringr| 3449|0.817%|\n",
      "|        plyr| 3436|0.814%|\n",
      "|    magrittr| 3265|0.774%|\n",
      "|      digest| 3223|0.764%|\n",
      "|    reshape2| 3205| 0.76%|\n",
      "|RColorBrewer| 3046|0.722%|\n",
      "|      scales| 3007|0.713%|\n",
      "+------------+-----+------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import udf\n",
    "\n",
    "derive_perc = udf(lambda x: str(round(x * 100, 3)) + \"%\")\n",
    "# or \n",
    "# @udf\n",
    "# def derive_perc(x):\n",
    "#     return(str(round(x * 100, 3)) + \"%\")\n",
    "\n",
    "package_count = package_count.withColumn(\"perc\", derive_perc(package_count['count'] / DT3.count()))\n",
    "\n",
    "package_count.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+------+\n",
      "|package|count|  perc|\n",
      "+-------+-----+------+\n",
      "|     DT|   97|0.023%|\n",
      "+-------+-----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "package_count.filter(package_count.package == 'DT').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. Using Spark SQL"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can build a view with the Spark DataFrame, then we can SQL syntax to further process our data.\n",
    "\n",
    "You may notice there are two ways to build a view,\n",
    "\n",
    "- `DF.createGlobalTempView` (or `DF.createOrReplaceGlobalTempView`): create a global temporary view\n",
    "- `DF.createTempView` (or `DF.createOrReplaceTempView`): create a local temporary view\n",
    "\n",
    "The main difference between them is the lifetime. The lifetime of a global temporary view is tied to the Spark application, while lifetime of a local temporary view is tied to the SparkSession."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creates or replaces a local temporary view with a DataFrame.\n",
    "# The lifetime of this temporary table is tied to the SparkSession that was used to create this DataFrame.\n",
    "\n",
    "package_count.createOrReplaceTempView(\"package_count_sql_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(perc=u'0.023%')]\n"
     ]
    }
   ],
   "source": [
    "# Basic Spark SQL Query - 1\n",
    "query_result = sqlContext.sql(\"select perc \\\n",
    "                               from package_count_sql_table \\\n",
    "                               where package = 'DT'\")\n",
    "\n",
    "print(query_result.collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+------+\n",
      "|package|count|  perc|\n",
      "+-------+-----+------+\n",
      "|   slam| 1006|0.238%|\n",
      "|     sp| 1020|0.242%|\n",
      "|  shiny| 1041|0.247%|\n",
      "|  tidyr| 1042|0.247%|\n",
      "|plotrix| 1066|0.253%|\n",
      "+-------+-----+------+\n",
      "only showing top 5 rows\n",
      "\n",
      "None\n"
     ]
    }
   ],
   "source": [
    "# Basic Spark SQL Query - 2\n",
    "query_result = sqlContext.sql(\"select * \\\n",
    "                                from package_count_sql_table \\\n",
    "                                where count > 1000 \\\n",
    "                                order by count asc\")\n",
    "print(query_result.show(5))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['slam:0.238%',\n",
       " 'sp:0.242%',\n",
       " 'shiny:0.247%',\n",
       " 'tidyr:0.247%',\n",
       " 'plotrix:0.253%',\n",
       " 'wordcloud:0.254%',\n",
       " 'rgl:0.257%',\n",
       " 'markdown:0.261%',\n",
       " 'irlba:0.27%',\n",
       " 'pkgmaker:0.27%']"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Use the Spark RDD way to process the results from Spark SQL query result\n",
    "query_result.rdd.map(lambda x:x['package'] + \":\" + x['perc']).take(10)"
   ]
  }
 ],
 "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.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}