spark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'C:\\\\spark'"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# bind to spark and initialize it\n",
    "\n",
    "import findspark\n",
    "findspark.init()\n",
    "import pyspark\n",
    "#  introduce most of the functionality of PySpark dataframes and SQL functionality\n",
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# start Spark Session\n",
    "conf = pyspark.SparkConf().setAppName('appName').setMaster('local')\n",
    "sc = pyspark.SparkContext(conf=conf)\n",
    "spark = SparkSession(sc)\n",
    "\n",
    "# check right spark\n",
    "findspark.find()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# full source in pys_sql.py\n",
    "from IPython.core.magic import register_line_cell_magic\n",
    "\n",
    "# Configuration parameters\n",
    "max_show_lines = 50         # Limit on the number of lines to show with %sql_show and %sql_display\n",
    "detailed_explain = True     # Set to False if you want to see only the physical plan when running explain\n",
    "\n",
    "\n",
    "@register_line_cell_magic\n",
    "def sql(line, cell=None):\n",
    "    \"Return a Spark DataFrame for lazy evaluation of the sql. Use: %sql or %%sql\"\n",
    "    val = cell if cell is not None else line \n",
    "    return spark.sql(val)\n",
    "\n",
    "@register_line_cell_magic\n",
    "def sql_show(line, cell=None):\n",
    "    \"Execute sql and show the first max_show_lines lines. Use: %sql_show or %%sql_show\"\n",
    "    val = cell if cell is not None else line \n",
    "    return spark.sql(val).show(max_show_lines) \n",
    "\n",
    "@register_line_cell_magic\n",
    "def sql_display(line, cell=None):\n",
    "    \"\"\"Execute sql and convert results to Pandas DataFrame for pretty display or further processing.\n",
    "    Use: %sql_display or %%sql_display\"\"\"\n",
    "    val = cell if cell is not None else line \n",
    "    return spark.sql(val).limit(max_show_lines).toPandas() \n",
    "\n",
    "@register_line_cell_magic\n",
    "def sql_explain(line, cell=None):\n",
    "    \"Display the execution plan of the sql. Use: %sql_explain or %%sql_explain\"\n",
    "    val = cell if cell is not None else line \n",
    "    return spark.sql(val).explain(detailed_explain)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Some small Demos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>(2 + 1)</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <td>0</td>\n",
       "      <td>3</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   (2 + 1)\n",
       "0        3"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%sql_display\n",
    "--#this is a cell in sql  \n",
    "select 2+1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[1, 4, 9]"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# example spark operation\n",
    "nums = sc.parallelize(range(1,4))\n",
    "nums.map(lambda x: x*x).collect()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [],
   "source": [
    "# # to stop the PySpark Session \n",
    "# sc.stop()\n",
    "#"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[_c0: string, body: string, author: string, created_utc: string, subreddit_id: string, link_id: string, parent_id: string, score: string, retrieved_on: string, controversiality: string, gilded: string, id: string, subreddit: string]"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# sc = pyspark.SparkContext(conf=conf)\n",
    "# spark = SparkSession(sc)\n",
    "\n",
    "# File location and type\n",
    "file_location = \"C:\\\\Users\\\\Skyler\\\\Downloads\\\\worldnews.csv\"\n",
    "file_type = \"csv\"\n",
    "\n",
    "# CSV options\n",
    "infer_schema = \"true\"\n",
    "first_row_is_header = \"true\"\n",
    "delimiter = \",\"\n",
    "\n",
    "# The applied options are for CSV files. For other file types, these will be ignored.\n",
    "products = spark.read.format(file_type) \\\n",
    "  .option(\"inferSchema\", infer_schema) \\\n",
    "  .option(\"header\", first_row_is_header) \\\n",
    "  .option(\"sep\", delimiter) \\\n",
    "  .load(file_location)\n",
    "\n",
    "display(products)\n",
    "products.createOrReplaceTempView(\"rc1\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>_c0</th>\n",
       "      <th>body</th>\n",
       "      <th>author</th>\n",
       "      <th>created_utc</th>\n",
       "      <th>subreddit_id</th>\n",
       "      <th>link_id</th>\n",
       "      <th>parent_id</th>\n",
       "      <th>score</th>\n",
       "      <th>retrieved_on</th>\n",
       "      <th>controversiality</th>\n",
       "      <th>gilded</th>\n",
       "      <th>id</th>\n",
       "      <th>subreddit</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>Slavs were considered Untermensch (less than h...</td>\n",
       "      <td>Say_less_fam</td>\n",
       "      <td>1560902706</td>\n",
       "      <td>t5_2qh13</td>\n",
       "      <td>t3_c2348g</td>\n",
       "      <td>t1_eriaihr</td>\n",
       "      <td>9</td>\n",
       "      <td>1570234835</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>erio2yn</td>\n",
       "      <td>worldnews</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>[removed]</td>\n",
       "      <td>[deleted]</td>\n",
       "      <td>1560125984</td>\n",
       "      <td>t5_2qh13</td>\n",
       "      <td>t3_bykehl</td>\n",
       "      <td>t1_eqk7hl9</td>\n",
       "      <td>1</td>\n",
       "      <td>1569308329</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>eqkloob</td>\n",
       "      <td>worldnews</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>2</td>\n",
       "      <td>2</td>\n",
       "      <td>It's been proven that there were several gassi...</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>3</td>\n",
       "      <td>It's never been proven that Assad gassed his o...</td>\n",
       "      <td>timelow</td>\n",
       "      <td>1560644549</td>\n",
       "      <td>t5_2qh13</td>\n",
       "      <td>t3_c11ea8</td>\n",
       "      <td>t1_eragun2</td>\n",
       "      <td>3</td>\n",
       "      <td>1570058384</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>erahzo3</td>\n",
       "      <td>worldnews</td>\n",
       "      <td>None</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>4</td>\n",
       "      <td>3</td>\n",
       "      <td>Heading towards?</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                                 _c0  \\\n",
       "0                                                  0   \n",
       "1                                                  1   \n",
       "2                                                  2   \n",
       "3  It's never been proven that Assad gassed his o...   \n",
       "4                                                  3   \n",
       "\n",
       "                                                body        author  \\\n",
       "0  Slavs were considered Untermensch (less than h...  Say_less_fam   \n",
       "1                                          [removed]     [deleted]   \n",
       "2  It's been proven that there were several gassi...          None   \n",
       "3                                            timelow    1560644549   \n",
       "4                                   Heading towards?          None   \n",
       "\n",
       "  created_utc subreddit_id     link_id   parent_id       score retrieved_on  \\\n",
       "0  1560902706     t5_2qh13   t3_c2348g  t1_eriaihr           9   1570234835   \n",
       "1  1560125984     t5_2qh13   t3_bykehl  t1_eqk7hl9           1   1569308329   \n",
       "2        None         None        None        None        None         None   \n",
       "3    t5_2qh13    t3_c11ea8  t1_eragun2           3  1570058384            0   \n",
       "4        None         None        None        None        None         None   \n",
       "\n",
       "  controversiality   gilded         id  subreddit  \n",
       "0                0        0    erio2yn  worldnews  \n",
       "1                0        0    eqkloob  worldnews  \n",
       "2             None     None       None       None  \n",
       "3                0  erahzo3  worldnews       None  \n",
       "4             None     None       None       None  "
      ]
     },
     "execution_count": 77,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%sql_display\n",
    "\n",
    "select * from rc1 limit 5"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- author: string (nullable = true)\n",
      " |-- author_flair_css_class: string (nullable = true)\n",
      " |-- author_flair_text: string (nullable = true)\n",
      " |-- body: string (nullable = true)\n",
      " |-- controversiality: long (nullable = true)\n",
      " |-- created_utc: long (nullable = true)\n",
      " |-- distinguished: string (nullable = true)\n",
      " |-- edited: boolean (nullable = true)\n",
      " |-- gilded: long (nullable = true)\n",
      " |-- id: string (nullable = true)\n",
      " |-- link_id: string (nullable = true)\n",
      " |-- parent_id: string (nullable = true)\n",
      " |-- retrieved_on: long (nullable = true)\n",
      " |-- score: long (nullable = true)\n",
      " |-- stickied: boolean (nullable = true)\n",
      " |-- subreddit: string (nullable = true)\n",
      " |-- subreddit_id: string (nullable = true)\n",
      " |-- ups: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pathlib import Path\n",
    "\n",
    "jsonpath=\n",
    "\n",
    "people = spark.read.json()\n",
    "# people.printSchema()\n",
    "people.createOrReplaceTempView(\"rcjson\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>author</th>\n",
       "      <th>author_flair_css_class</th>\n",
       "      <th>author_flair_text</th>\n",
       "      <th>body</th>\n",
       "      <th>controversiality</th>\n",
       "      <th>created_utc</th>\n",
       "      <th>distinguished</th>\n",
       "      <th>edited</th>\n",
       "      <th>gilded</th>\n",
       "      <th>id</th>\n",
       "      <th>link_id</th>\n",
       "      <th>parent_id</th>\n",
       "      <th>retrieved_on</th>\n",
       "      <th>score</th>\n",
       "      <th>stickied</th>\n",
       "      <th>subreddit</th>\n",
       "      <th>subreddit_id</th>\n",
       "      <th>ups</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <td>0</td>\n",
       "      <td>frjo</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>A look at Vietnam and Mexico exposes the myth ...</td>\n",
       "      <td>0</td>\n",
       "      <td>1134365188</td>\n",
       "      <td>None</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>c13</td>\n",
       "      <td>t3_17863</td>\n",
       "      <td>t3_17863</td>\n",
       "      <td>1473738411</td>\n",
       "      <td>2</td>\n",
       "      <td>False</td>\n",
       "      <td>reddit.com</td>\n",
       "      <td>t5_6</td>\n",
       "      <td>2</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>1</td>\n",
       "      <td>zse7zse</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>The site states \"What can I use it for? Meetin...</td>\n",
       "      <td>0</td>\n",
       "      <td>1134365725</td>\n",
       "      <td>None</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>c14</td>\n",
       "      <td>t3_17866</td>\n",
       "      <td>t3_17866</td>\n",
       "      <td>1473738411</td>\n",
       "      <td>1</td>\n",
       "      <td>False</td>\n",
       "      <td>reddit.com</td>\n",
       "      <td>t5_6</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>2</td>\n",
       "      <td>[deleted]</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>Jython related topics by Frank Wierzbicki</td>\n",
       "      <td>0</td>\n",
       "      <td>1134366848</td>\n",
       "      <td>None</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>c15</td>\n",
       "      <td>t3_17869</td>\n",
       "      <td>t3_17869</td>\n",
       "      <td>1473738411</td>\n",
       "      <td>0</td>\n",
       "      <td>False</td>\n",
       "      <td>reddit.com</td>\n",
       "      <td>t5_6</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>3</td>\n",
       "      <td>[deleted]</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>[deleted]</td>\n",
       "      <td>0</td>\n",
       "      <td>1134367660</td>\n",
       "      <td>None</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>c16</td>\n",
       "      <td>t3_17870</td>\n",
       "      <td>t3_17870</td>\n",
       "      <td>1473738411</td>\n",
       "      <td>1</td>\n",
       "      <td>False</td>\n",
       "      <td>reddit.com</td>\n",
       "      <td>t5_6</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <td>4</td>\n",
       "      <td>rjoseph</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>Saft is by far the best extension you could ta...</td>\n",
       "      <td>0</td>\n",
       "      <td>1134367754</td>\n",
       "      <td>None</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>c17</td>\n",
       "      <td>t3_17817</td>\n",
       "      <td>t3_17817</td>\n",
       "      <td>1473738411</td>\n",
       "      <td>1</td>\n",
       "      <td>False</td>\n",
       "      <td>reddit.com</td>\n",
       "      <td>t5_6</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "      author author_flair_css_class author_flair_text  \\\n",
       "0       frjo                   None              None   \n",
       "1    zse7zse                   None              None   \n",
       "2  [deleted]                   None              None   \n",
       "3  [deleted]                   None              None   \n",
       "4    rjoseph                   None              None   \n",
       "\n",
       "                                                body  controversiality  \\\n",
       "0  A look at Vietnam and Mexico exposes the myth ...                 0   \n",
       "1  The site states \"What can I use it for? Meetin...                 0   \n",
       "2          Jython related topics by Frank Wierzbicki                 0   \n",
       "3                                          [deleted]                 0   \n",
       "4  Saft is by far the best extension you could ta...                 0   \n",
       "\n",
       "   created_utc distinguished  edited  gilded   id   link_id parent_id  \\\n",
       "0   1134365188          None   False       0  c13  t3_17863  t3_17863   \n",
       "1   1134365725          None   False       0  c14  t3_17866  t3_17866   \n",
       "2   1134366848          None   False       0  c15  t3_17869  t3_17869   \n",
       "3   1134367660          None   False       0  c16  t3_17870  t3_17870   \n",
       "4   1134367754          None   False       0  c17  t3_17817  t3_17817   \n",
       "\n",
       "   retrieved_on  score  stickied   subreddit subreddit_id  ups  \n",
       "0    1473738411      2     False  reddit.com         t5_6    2  \n",
       "1    1473738411      1     False  reddit.com         t5_6    1  \n",
       "2    1473738411      0     False  reddit.com         t5_6    0  \n",
       "3    1473738411      1     False  reddit.com         t5_6    1  \n",
       "4    1473738411      1     False  reddit.com         t5_6    1  "
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%sql_display\n",
    "\n",
    "select * from rcjson limit 5"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "ename": "AnalysisException",
     "evalue": "Table `reddit` already exists.;",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mPy4JJavaError\u001b[0m                             Traceback (most recent call last)",
      "\u001b[1;32mC:\\spark\\python\\pyspark\\sql\\utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     97\u001b[0m         \u001b[1;32mtry\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 98\u001b[1;33m             \u001b[1;32mreturn\u001b[0m \u001b[0mf\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;33m*\u001b[0m\u001b[0ma\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;33m**\u001b[0m\u001b[0mkw\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     99\u001b[0m         \u001b[1;32mexcept\u001b[0m \u001b[0mpy4j\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mprotocol\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mPy4JJavaError\u001b[0m \u001b[1;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mC:\\spark\\python\\lib\\py4j-0.10.8.1-src.zip\\py4j\\protocol.py\u001b[0m in \u001b[0;36mget_return_value\u001b[1;34m(answer, gateway_client, target_id, name)\u001b[0m\n\u001b[0;32m    327\u001b[0m                     \u001b[1;34m\"An error occurred while calling {0}{1}{2}.\\n\"\u001b[0m\u001b[1;33m.\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 328\u001b[1;33m                     format(target_id, \".\", name), value)\n\u001b[0m\u001b[0;32m    329\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mPy4JJavaError\u001b[0m: An error occurred while calling o177.saveAsTable.\n: org.apache.spark.sql.AnalysisException: Table `reddit` already exists.;\r\n\tat org.apache.spark.sql.DataFrameWriter.saveAsTable(DataFrameWriter.scala:573)\r\n\tat org.apache.spark.sql.DataFrameWriter.saveAsTable(DataFrameWriter.scala:498)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\r\n\tat sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)\r\n\tat sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)\r\n\tat java.lang.reflect.Method.invoke(Unknown Source)\r\n\tat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\r\n\tat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\r\n\tat py4j.Gateway.invoke(Gateway.java:282)\r\n\tat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\r\n\tat py4j.commands.CallCommand.execute(CallCommand.java:79)\r\n\tat py4j.GatewayConnection.run(GatewayConnection.java:238)\r\n\tat java.lang.Thread.run(Unknown Source)\r\n",
      "\nDuring handling of the above exception, another exception occurred:\n",
      "\u001b[1;31mAnalysisException\u001b[0m                         Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-16-d4846b5843c1>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m\u001b[0m\n\u001b[0;32m      1\u001b[0m \u001b[1;31m# write database to persistant form\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m      2\u001b[0m \u001b[0mpermanent_table_name\u001b[0m \u001b[1;33m=\u001b[0m \u001b[1;34m\"reddit\"\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m----> 3\u001b[1;33m \u001b[0mproducts\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mwrite\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mformat\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"parquet\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msaveAsTable\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpermanent_table_name\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;32mC:\\spark\\python\\pyspark\\sql\\readwriter.py\u001b[0m in \u001b[0;36msaveAsTable\u001b[1;34m(self, name, format, mode, partitionBy, **options)\u001b[0m\n\u001b[0;32m    785\u001b[0m         \u001b[1;32mif\u001b[0m \u001b[0mformat\u001b[0m \u001b[1;32mis\u001b[0m \u001b[1;32mnot\u001b[0m \u001b[1;32mNone\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    786\u001b[0m             \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mformat\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mformat\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 787\u001b[1;33m         \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jwrite\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msaveAsTable\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mname\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    788\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    789\u001b[0m     \u001b[1;33m@\u001b[0m\u001b[0msince\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;36m1.4\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mC:\\spark\\python\\lib\\py4j-0.10.8.1-src.zip\\py4j\\java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[1;34m(self, *args)\u001b[0m\n\u001b[0;32m   1284\u001b[0m         \u001b[0manswer\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1285\u001b[0m         return_value = get_return_value(\n\u001b[1;32m-> 1286\u001b[1;33m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[0;32m   1287\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m   1288\u001b[0m         \u001b[1;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[1;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32mC:\\spark\\python\\pyspark\\sql\\utils.py\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m    100\u001b[0m             \u001b[0mconverted\u001b[0m \u001b[1;33m=\u001b[0m \u001b[0mconvert_exception\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0me\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mjava_exception\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    101\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[1;32mnot\u001b[0m \u001b[0misinstance\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mconverted\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mUnknownException\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 102\u001b[1;33m                 \u001b[1;32mraise\u001b[0m \u001b[0mconverted\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    103\u001b[0m             \u001b[1;32melse\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    104\u001b[0m                 \u001b[1;32mraise\u001b[0m\u001b[1;33m\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mAnalysisException\u001b[0m: Table `reddit` already exists.;"
     ]
    }
   ],
   "source": [
    "# write database to persistant form\n",
    "permanent_table_name = \"reddit\"\n",
    "products.write.format(\"parquet\").saveAsTable(permanent_table_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {},
   "outputs": [],
   "source": [
    "# sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here is the first bit of heavy lifting where PySpark excels. It can directly read the bz2 files and parse the json."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+--------------------+-----------+-----+\n",
      "|       author|                body|created_utc|score|\n",
      "+-------------+--------------------+-----------+-----+\n",
      "|procrastitron|Thanks for the li...| 1164931204|   12|\n",
      "|       luke_s|Even today some o...| 1164931222|    6|\n",
      "+-------------+--------------------+-----------+-----+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df=spark.read.json(\"I:\\\\reddit_data\\\\reddit\\\\2006\\\\RC_2006-12.bz2\")\n",
    "df.select(\"author\",\"body\",\"created_utc\",\"score\").show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}