SQL-Cloudant-connector-Python-example.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Python example using Spark SQL over Cloudant as a source\n",
    "\n",
    "This sample notebook is written in Python and expects the Python 2.7.5 runtime. Make sure the kernel is started and you are connect to it when executing this notebook.\n",
    "The data source for this example can be found at: http://examples.cloudant.com/crimes/. Replicate the database into your own Cloudant account before you execute this script.\n",
    "\n",
    "This Python notebook showcases how to use the SQL-Cloudant connector. This code reads Cloudant data, creates a DataFrame from the Cloudant data, filters that data down to only crime incidents with the nature code for a public disturbance, and then writes those 7 documents to another Cloudant database."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Work with SparkSession\n",
    "\n",
    "Import and initialize SparkSession."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Work with a Cloudant database\n",
    "\n",
    "A Dataframe object can be created directly from a Cloudant database. To configure the database as source, pass these options:\n",
    "\n",
    "1 - package name that provides the classes (like CloudantDataSource) implemented in the connector to extend BaseRelation. For the SQL-Cloudant connector this will be org.apache.bahir.cloudant\n",
    "\n",
    "2 - cloudant.host parameter to pass the Cloudant account name\n",
    "\n",
    "3 - cloudant.user parameter to pass the Cloudant user name\n",
    "\n",
    "4 - cloudant.password parameter to pass the Cloudant account password\n",
    "\n",
    "5 - the database to load\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "cloudantdata = spark.read.format(\"org.apache.bahir.cloudant\")\\\n",
    "    .option(\"cloudant.host\",\"xxxxxxxxxxxxxx-bluemix.cloudant.com\")\\\n",
    "    .option(\"cloudant.username\", \"xxxxxxxxxxxxxxxx\")\\\n",
    "    .option(\"cloudant.password\",\"xxxxxxxxxxxxxxxxxxxxx\")\\\n",
    "    .load(\"crimes\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Work with a Dataframe\n",
    "\n",
    "At this point, all transformations and functions should behave as specified with Spark SQL. (http://spark.apache.org/sql/)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This code prints the schema and a record count\n",
    "cloudantdata.printSchema()\n",
    "cloudantdata.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This code displays the values of the naturecode field\n",
    "cloudantdata.select(\"properties.naturecode\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This code filters the data to just those records with a naturecode of \"DISTRB\", and then displays that data\n",
    "disturbDF = cloudantdata.filter(\"properties.naturecode = 'DISTRB'\")\n",
    "disturbDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# This code writes the filtered data to a Cloudant database called crimes_filtered. \n",
    "# To avoid error, the Cloudant database must already exist\n",
    "disturbDF.select(\"properties\").write.format(\"org.apache.bahir.cloudant\")\\\n",
    "     .option(\"cloudant.host\",\"xxxxxxxxxxxxxx-bluemix.cloudant.com\")\\\n",
    "     .option(\"cloudant.username\", \"xxxxxxxxxxxxxxxx\")\\\n",
    "     .option(\"cloudant.password\",\"xxxxxxxxxxxxxxxxxxxxx\")\\\n",
    "     .option(\"createDBOnSave\", \"true\")\\\n",
    "     .save(\"crimes_filtered\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Next, you'll see how to create a visualization of the crimes data. \n",
    "# First, this line creates a DataFrame containing all of the naturecodes and a count of the crime incidents for each code.\n",
    "reducedValue = cloudantdata.groupBy(\"properties.naturecode\").count()\n",
    "reducedValue.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Generate visualizations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# This line imports two Python modules. The pprint module helps to produce pretty representations of data structures, \n",
    "# and the counter subclass from the collections module helps to count hashable objects.\n",
    "import pprint\n",
    "from collections import Counter"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# This line imports PySpark classes for Spark SQL and DataFrames.\n",
    "from pyspark.sql import *\n",
    "from pyspark.sql.functions import udf, asc, desc\n",
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql.types import IntegerType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# This line converts an Apache Spark DataFrame to a Panda DataFrame, and also sorts the DataFrame by count first, \n",
    "# and then by naturecode second in order to produce a sorted graph later.\n",
    "import pandas as pd\n",
    "pandaDF = reducedValue.orderBy(desc(\"count\"), asc(\"naturecode\")).toPandas()\n",
    "print(pandaDF)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# This is needed to actually see the plots\n",
    "%matplotlib inline\n",
    "\n",
    "# This line imports matplotlib.pyplot which is a collection of command style functions that make matplotlib work like MATLAB\n",
    "import matplotlib.pyplot as plt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# These lines assign the data to the values and labels objects.\n",
    "values = pandaDF['count']\n",
    "labels = pandaDF['naturecode']\n",
    "\n",
    "# These lines provide the format for the plot.\n",
    "plt.gcf().set_size_inches(16, 12, forward=True)\n",
    "plt.title('Number of crimes by type')\n",
    "\n",
    "# These lines specify that the plot should display as a horizontal bar chart with values being for the x axis \n",
    "# and labels for the y axis\n",
    "plt.barh(range(len(values)), values)\n",
    "plt.yticks(range(len(values)), labels)\n",
    "\n",
    "# This last line displays the plot\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2 with Spark 2.1",
   "language": "python",
   "name": "python2-spark21"
  },
  "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.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}