DF 1.1.1 - Built-in Data Sources.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.context.SparkContext at 0x7f8bc85c7790>"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "!rm ./metastore_db/*.lck\n",
    "examples_folder = \"/usr/local/share/spark/examples/src/main/resources/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "sqlc = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Parquet Files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------------+----------------+\n",
      "|  name|favorite_color|favorite_numbers|\n",
      "+------+--------------+----------------+\n",
      "|Alyssa|          null|  [3, 9, 15, 20]|\n",
      "|   Ben|           red|              []|\n",
      "+------+--------------+----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = sqlc.read.load(examples_folder + \"users.parquet\")\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!rm -rf namesAndFavColors.parquet/\n",
    "df.select(\"name\", \"favorite_color\").write.save(\"namesAndFavColors.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "total 4\r\n",
      "-rw------- 1 dvgodoy dvgodoy 509 Set 25 17:04 part-r-00000-876c04e5-04ce-44a9-a451-ec2074838894.snappy.parquet\r\n",
      "-rw------- 1 dvgodoy dvgodoy   0 Set 25 17:04 _SUCCESS\r\n"
     ]
    }
   ],
   "source": [
    "!ls -l namesAndFavColors.parquet/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[name: string, favorite_color: string, favorite_numbers: array<int>]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sqlc.read.format(\"org.apache.spark.sql.parquet\").load(\"users.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[name: string, favorite_color: string, favorite_numbers: array<int>]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sqlc.read.format(\"parquet\").load(\"users.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[name: string, favorite_color: string, favorite_numbers: array<int>]"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sqlc.read.parquet(\"users.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## JSON Files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------+\n",
      "| age|   name|\n",
      "+----+-------+\n",
      "|null|Michael|\n",
      "|  30|   Andy|\n",
      "|  19| Justin|\n",
      "+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_json = sqlc.read.format(\"json\").load(examples_folder + \"people.json\")\n",
    "df_json.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- age: long (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_json.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'{\"name\":\"Michael\"}',\n",
       " u'{\"age\":30,\"name\":\"Andy\"}',\n",
       " u'{\"age\":19,\"name\":\"Justin\"}']"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "json_strings = df_json.toJSON()\n",
    "json_strings.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!rm -rf namesAndAges.parquet/\n",
    "df_json.select(\"name\", \"age\").write.format(\"parquet\").save(\"namesAndAges.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "total 1\r\n",
      "-rw------- 1 dvgodoy dvgodoy 522 Set 25 17:09 part-r-00000-931a292f-ba66-4df2-8f10-87a8abb42289.snappy.parquet\r\n",
      "-rw------- 1 dvgodoy dvgodoy   0 Set 25 17:09 _SUCCESS\r\n"
     ]
    }
   ],
   "source": [
    "!ls -l namesAndAges.parquet"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!rm -rf namesAndAgesCoalesced.parquet/\n",
    "df_json.select(\"name\", \"age\").coalesce(1).write.format(\"parquet\").save(\"namesAndAgesCoalesced.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "total 1\r\n",
      "-rw------- 1 dvgodoy dvgodoy 522 Set 25 17:10 part-r-00000-038eb1d1-1907-4923-a0ea-d60dc1401151.snappy.parquet\r\n",
      "-rw------- 1 dvgodoy dvgodoy   0 Set 25 17:10 _SUCCESS\r\n"
     ]
    }
   ],
   "source": [
    "!ls -l namesAndAgesCoalesced.parquet"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "ename": "AnalysisException",
     "evalue": "u'path file:/media/dvgodoy/FILES/DataScienceRetreat/SparkClass/namesAndAges.parquet already exists.;'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mAnalysisException\u001b[0m                         Traceback (most recent call last)",
      "\u001b[1;32m<ipython-input-30-5d4993300a51>\u001b[0m in \u001b[0;36m<module>\u001b[1;34m()\u001b[0m\n\u001b[1;32m----> 1\u001b[1;33m \u001b[0mdf_json\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mwrite\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mmode\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"error\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mparquet\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m\"namesAndAges.parquet\"\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[1;32m/usr/local/share/spark/python/pyspark/sql/readwriter.pyc\u001b[0m in \u001b[0;36mparquet\u001b[1;34m(self, path, mode, partitionBy, compression)\u001b[0m\n\u001b[0;32m    615\u001b[0m             \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mpartitionBy\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpartitionBy\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    616\u001b[0m         \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_set_opts\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mcompression\u001b[0m\u001b[1;33m=\u001b[0m\u001b[0mcompression\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m--> 617\u001b[1;33m         \u001b[0mself\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0m_jwrite\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mparquet\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0mpath\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m    618\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    619\u001b[0m     \u001b[1;33m@\u001b[0m\u001b[0msince\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;36m1.6\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;32m/usr/local/share/spark/python/lib/py4j-0.10.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    931\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[0m\n\u001b[0;32m    932\u001b[0m         return_value = get_return_value(\n\u001b[1;32m--> 933\u001b[1;33m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[0;32m    934\u001b[0m \u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m    935\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[0m\n",
      "\u001b[1;32m/usr/local/share/spark/python/pyspark/sql/utils.pyc\u001b[0m in \u001b[0;36mdeco\u001b[1;34m(*a, **kw)\u001b[0m\n\u001b[0;32m     67\u001b[0m                                              e.java_exception.getStackTrace()))\n\u001b[0;32m     68\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'org.apache.spark.sql.AnalysisException: '\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[1;32m---> 69\u001b[1;33m                 \u001b[1;32mraise\u001b[0m \u001b[0mAnalysisException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m': '\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0m\u001b[0;32m     70\u001b[0m             \u001b[1;32mif\u001b[0m \u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0mstartswith\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m'org.apache.spark.sql.catalyst.analysis'\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m:\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n\u001b[0;32m     71\u001b[0m                 \u001b[1;32mraise\u001b[0m \u001b[0mAnalysisException\u001b[0m\u001b[1;33m(\u001b[0m\u001b[0ms\u001b[0m\u001b[1;33m.\u001b[0m\u001b[0msplit\u001b[0m\u001b[1;33m(\u001b[0m\u001b[1;34m': '\u001b[0m\u001b[1;33m,\u001b[0m \u001b[1;36m1\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m[\u001b[0m\u001b[1;36m1\u001b[0m\u001b[1;33m]\u001b[0m\u001b[1;33m,\u001b[0m \u001b[0mstackTrace\u001b[0m\u001b[1;33m)\u001b[0m\u001b[1;33m\u001b[0m\u001b[0m\n",
      "\u001b[1;31mAnalysisException\u001b[0m: u'path file:/media/dvgodoy/FILES/DataScienceRetreat/SparkClass/namesAndAges.parquet already exists.;'"
     ]
    }
   ],
   "source": [
    "df_json.write.mode(\"error\").parquet(\"namesAndAges.parquet\")\n",
    "# This IS supposed to throw an exception, as it is trying to write an already existent file!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df_json.write.mode(\"overwrite\").parquet(\"namesAndAges.parquet\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## SQL on Files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+------+\n",
      "|age|  name|\n",
      "+---+------+\n",
      "| 19|Justin|\n",
      "+---+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = sqlc.sql(\"SELECT * FROM json.`people.json` WHERE age = 19\")\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## MySQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Run this commands in MYSQL\n",
    "\n",
    "# CREATE TABLE users (user_id int PRIMARY KEY, fname text, lname text);\n",
    "# INSERT INTO users (user_id,  fname, lname) VALUES (1, 'john', 'smith');\n",
    "# INSERT INTO users (user_id,  fname, lname) VALUES (2, 'john', 'doe');\n",
    "# INSERT INTO users (user_id,  fname, lname) VALUES (3, 'john', 'smith');"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df_mysql = sqlc.read.format(\"jdbc\") \\\n",
    "                .option(\"url\", \"jdbc:mysql://localhost:3306/test\") \\\n",
    "                .option(\"driver\", \"com.mysql.jdbc.Driver\") \\\n",
    "                .option(\"dbtable\", \"users\") \\\n",
    "                .option(\"user\", \"root\") \\\n",
    "                .option(\"password\", \"\").load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+-----+\n",
      "|user_id|fname|lname|\n",
      "+-------+-----+-----+\n",
      "|   1744| john|  doe|\n",
      "|   1745| john|smith|\n",
      "|   1746| john|smith|\n",
      "+-------+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_mysql.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- user_id: integer (nullable = false)\n",
      " |-- fname: string (nullable = true)\n",
      " |-- lname: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_mysql.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df_mysql.registerTempTable(\"users\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(user_id=1744, fname=u'john', lname=u'doe'),\n",
       " Row(user_id=1745, fname=u'john', lname=u'smith'),\n",
       " Row(user_id=1746, fname=u'john', lname=u'smith')]"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sqlc.sql(\"select * from users\").collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "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": 0
}