Import-to-Cloudant.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook is intended for Python 2 with Spark 2.0. It uses SparkSession to load a CSV file stored in Bluemix object storage into a dataframe, filters that data, then writes the filtered data to a previoulsy created Cloudant database. This example notebook loads a CSV file containing Child Care providers in Massachusetts downloaded from https://data.mass.gov/Education/Program-list-for-Child-Care-Search-1-15-2015/cb6m-ccic\n",
    "\n",
    "This first cell simply verifies the version of Spark you are using."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "spark.version"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cell 2: Replace the contents of the first cell by following these steps:\n",
    "1. Displaying the Files slide out panel.\n",
    "2. Select the Insert to code menu for your file, and select Insert Credentials.\n",
    "3. Replace the name of the inserted array with credentials_621 as referenced in the rest of the code. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# @hidden_cell\n",
    "credentials_621 = {\n",
    "  'auth_url':'https://identity.open.softlayer.com',\n",
    "  'project':'object_storage_xxxxxxxx',\n",
    "  'project_id':'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx',\n",
    "  'region':'dallas',\n",
    "  'user_id':'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx',\n",
    "  'domain_id':'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx',\n",
    "  'domain_name':'xxxxxxxx',\n",
    "  'username':'member_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx',\n",
    "  'password':\"\"\"xxxxxxxxxxxxxxxx\"\"\",\n",
    "  'container':'CloudantSparkIntegration',\n",
    "  'tenantId':'undefined',\n",
    "  'filename':'Program_list_for_Child_Care_Search_1-15-2015.csv'\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cell 3: The following cell imports SparkSession from pyspark.sql. SparkSession is the entry point to programming Spark with the Dataset and DataFrame API.\n",
    "Next, the code defines a variable to set the credentials for authentication for the Bluemix Object Storage."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# @hidden_cell\n",
    "# This function is used to setup the access of Spark to your Object Storage. The definition contains your credentials.\n",
    "# You might want to remove those credentials before you share your notebook.\n",
    "def set_hadoop_config_with_credentials_620ad16a(name):\n",
    "    \"\"\"This function sets the Hadoop configuration so it is possible to\n",
    "    access data from Bluemix Object Storage using Spark\"\"\"\n",
    "\n",
    "    prefix = 'fs.swift.service.' + name\n",
    "    hconf = sc._jsc.hadoopConfiguration()\n",
    "    hconf.set(prefix + '.auth.url', credentials_621['auth_url']+'/v3/auth/tokens')\n",
    "    hconf.set(prefix + '.auth.endpoint.prefix', 'endpoints')\n",
    "    hconf.set(prefix + '.tenant', credentials_621['project_id'])\n",
    "    hconf.set(prefix + '.username', credentials_621['user_id'])\n",
    "    hconf.set(prefix + '.password', credentials_621['password'])\n",
    "    hconf.setInt(prefix + '.http.port', 8080)\n",
    "    hconf.set(prefix + '.region', credentials_621['region'])\n",
    "    hconf.setBoolean(prefix + '.public', False)\n",
    "\n",
    "# you can choose any name\n",
    "name = 'keystone'\n",
    "set_hadoop_config_with_credentials_620ad16a(name)\n",
    "\n",
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cell 4: The following cell reads the CSV file into a data frame, infers the schema, and then displays the first two entries."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "massdata = spark.read\\\n",
    "  .format('org.apache.spark.sql.execution.datasources.csv.CSVFileFormat')\\\n",
    "  .option('header', 'true')\\\n",
    "  .option('timestampFormat', 'MM/dd/yyyy')\\\n",
    "  .option('inferSchema', 'true')\\\n",
    "  .load('swift://' + credentials_621['container'] + '.' + name + '/' + credentials_621['filename'])\n",
    "massdata.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cell 5: The following cell prints the schema and a record count of the data frame contents."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "massdata.printSchema()\n",
    "massdata.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cell 6: The following cell displays the first 30 values in the Session1Name field. Notice that there are null values."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "massdata.select(\"Session1Name\").show(30)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cell 7: The following cell filters the data to just those facilities that have a specified Session1Name. Then it displays the first two entries and a count of the filtered data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sessiondata = massdata.filter(massdata.Session1Name.isNotNull())\n",
    "sessiondata.show(2)\n",
    "sessiondata.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cell 8: The following cell displays the first 30 values in the Session1Name field. Notice that there are NO null values."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sessiondata.select(\"Session1Name\").show(30)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Cell 9: The following cell writes the contents of the sessiondata data frame to a Cloudant database called child_care. Note: The Cloudant database MUST already exist."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sessiondata.write.format(\"com.cloudant.spark\") \\\n",
    "  .option(\"cloudant.host\",\"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx-bluemix.cloudant.com\") \\\n",
    "  .option(\"cloudant.username\",\"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx-bluemix\") \\\n",
    "  .option(\"cloudant.password\",\"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\") \\\n",
    "  .save(\"child_care\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2 with Spark 2.0",
   "language": "python",
   "name": "python2-spark20"
  },
  "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": 0
}