pyspark_opendap_rdd.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import sys \n",
    "\n",
    "# $SPARK_HOME set to the directory where you\n",
    "# have the source download unpacked\n",
    "spark_home = os.environ.get('SPARK_HOME', None)\n",
    "\n",
    "if not spark_home:\n",
    "    raise ValueError('SPARK_HOME environment variable is not set')\n",
    "\n",
    "sys.path.insert(0, os.path.join(spark_home, 'python'))\n",
    "\n",
    "sys.path.insert(0, os.path.join(spark_home, 'python/lib/py4j-0.10.3-src.zip'))                                      \n",
    "\n",
    "# executing shell.py will create 'sc' a SparkContext\n",
    "#execfile(os.path.join(spark_home, 'python/pyspark/shell.py'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "import pydap.client\n",
    "import requests"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sc = SparkContext(\"local\", \"Simple App\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "## Will return dataset details, not the data\n",
    "#opendap_url = 'http://opendap.cr.usgs.gov/opendap/hyrax/MCD12Q1.051/h01v09.ncml'\n",
    "## URL for grabbing actual data\n",
    "opendap_url = 'http://opendap.cr.usgs.gov/opendap/hyrax/MOD09Q1.005/h11v04.ncml.ascii?sur_refl_b01[683:1:683][2000:1:2010][2000:1:2010]'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "## Raises Exception, unable to parse token\n",
    "# dataset = pydap.client.open_url(opendap_url)\n",
    "response = requests.get(opendap_url)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Dataset: h11v04.ncml\n",
      "sur_refl_b01[0][0], 0.5359, 0.5657, 0.5258, 0.4739, 0.4648, 0.5087, 0.5234, 0.5234, 0.6214, 0.5793, 0.5374\n",
      "sur_refl_b01[0][1], 0.5759, 0.5414, 0.5338, 0.4489, 0.432, 0.5036, 0.6051, 0.6214, 0.5793, 0.455, 0.446\n",
      "sur_refl_b01[0][2], 0.4594, 0.5409, 0.6224, 0.6289, 0.6289, 0.6051, 0.5005, 0.5005, 0.3325, 0.3092, 0.3826\n",
      "sur_refl_b01[0][3], 0.5831, 0.5831, 0.6126, 0.6307, 0.6307, 0.5277, 0.612, 0.3325, 0.3092, 0.332, 0.4036\n",
      "sur_refl_b01[0][4], 0.6329, 0.6126, 0.6061, 0.5691, 0.5691, 0.612, 0.5841, 0.5841, 0.4301, 0.4782, 0.6271\n",
      "sur_refl_b01[0][5], 0.6092, 0.6092, 0.6152, 0.5691, 0.6297, 0.6035, 0.5052, 0.5052, 0.4782, 0.6518, 0.6913\n",
      "sur_refl_b01[0][6], 0.5712, 0.6152, 0.6248, 0.5686, 0.5686, 0.3522, 0.5088, 0.5912, 0.6316, 0.4729, 0.5756\n",
      "sur_refl_b01[0][7], 0.5737, 0.5737, 0.524, 0.5368, 0.3522, 0.5691, 0.5944, 0.4085, 0.4729, 0.5375, 0.5375\n",
      "sur_refl_b01[0][8], 0.5606, 0.5211, 0.5368, 0.5874, 0.5874, 0.343, 0.26, 0.26, 0.3825, 0.5794, 0.5863\n",
      "sur_refl_b01[0][9], 0.5883, 0.5591, 0.5263, 0.4325, 0.343, 0.4485, 0.626, 0.626, 0.5462, 0.3682, 0.3463\n",
      "sur_refl_b01[0][10], 0.5435, 0.5263, 0.4325, 0.4887, 0.4887, 0.5175, 0.626, 0.5475, 0.4101, 0.4592, 0.4203\n",
      "\n"
     ]
    }
   ],
   "source": [
    "for i in response.text.split(\"\\n\"):\n",
    "    print i"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# A single use/disposable function\n",
    "def dictorize(dataobj):\n",
    "    od, dkey = {}, ''\n",
    "    _dlist = dataobj.split('\\n')\n",
    "    for index, val in enumerate(_dlist):\n",
    "        if index == 0:\n",
    "            dkey = val.split(':')[1]\n",
    "            od[dkey] = {}\n",
    "        else:\n",
    "            _vlist = val.split(',')\n",
    "            _vind = len(_vlist) - 1\n",
    "            if _vlist[0]:\n",
    "                od[dkey][_vlist[0]] = [float(i) for i in _vlist[-_vind:]]\n",
    "    return od"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "simple_dataset = dictorize(response.text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{u' h11v04.ncml': {u'sur_refl_b01[0][0]': [0.5359,\n",
       "   0.5657,\n",
       "   0.5258,\n",
       "   0.4739,\n",
       "   0.4648,\n",
       "   0.5087,\n",
       "   0.5234,\n",
       "   0.5234,\n",
       "   0.6214,\n",
       "   0.5793,\n",
       "   0.5374],\n",
       "  u'sur_refl_b01[0][10]': [0.5435,\n",
       "   0.5263,\n",
       "   0.4325,\n",
       "   0.4887,\n",
       "   0.4887,\n",
       "   0.5175,\n",
       "   0.626,\n",
       "   0.5475,\n",
       "   0.4101,\n",
       "   0.4592,\n",
       "   0.4203],\n",
       "  u'sur_refl_b01[0][1]': [0.5759,\n",
       "   0.5414,\n",
       "   0.5338,\n",
       "   0.4489,\n",
       "   0.432,\n",
       "   0.5036,\n",
       "   0.6051,\n",
       "   0.6214,\n",
       "   0.5793,\n",
       "   0.455,\n",
       "   0.446],\n",
       "  u'sur_refl_b01[0][2]': [0.4594,\n",
       "   0.5409,\n",
       "   0.6224,\n",
       "   0.6289,\n",
       "   0.6289,\n",
       "   0.6051,\n",
       "   0.5005,\n",
       "   0.5005,\n",
       "   0.3325,\n",
       "   0.3092,\n",
       "   0.3826],\n",
       "  u'sur_refl_b01[0][3]': [0.5831,\n",
       "   0.5831,\n",
       "   0.6126,\n",
       "   0.6307,\n",
       "   0.6307,\n",
       "   0.5277,\n",
       "   0.612,\n",
       "   0.3325,\n",
       "   0.3092,\n",
       "   0.332,\n",
       "   0.4036],\n",
       "  u'sur_refl_b01[0][4]': [0.6329,\n",
       "   0.6126,\n",
       "   0.6061,\n",
       "   0.5691,\n",
       "   0.5691,\n",
       "   0.612,\n",
       "   0.5841,\n",
       "   0.5841,\n",
       "   0.4301,\n",
       "   0.4782,\n",
       "   0.6271],\n",
       "  u'sur_refl_b01[0][5]': [0.6092,\n",
       "   0.6092,\n",
       "   0.6152,\n",
       "   0.5691,\n",
       "   0.6297,\n",
       "   0.6035,\n",
       "   0.5052,\n",
       "   0.5052,\n",
       "   0.4782,\n",
       "   0.6518,\n",
       "   0.6913],\n",
       "  u'sur_refl_b01[0][6]': [0.5712,\n",
       "   0.6152,\n",
       "   0.6248,\n",
       "   0.5686,\n",
       "   0.5686,\n",
       "   0.3522,\n",
       "   0.5088,\n",
       "   0.5912,\n",
       "   0.6316,\n",
       "   0.4729,\n",
       "   0.5756],\n",
       "  u'sur_refl_b01[0][7]': [0.5737,\n",
       "   0.5737,\n",
       "   0.524,\n",
       "   0.5368,\n",
       "   0.3522,\n",
       "   0.5691,\n",
       "   0.5944,\n",
       "   0.4085,\n",
       "   0.4729,\n",
       "   0.5375,\n",
       "   0.5375],\n",
       "  u'sur_refl_b01[0][8]': [0.5606,\n",
       "   0.5211,\n",
       "   0.5368,\n",
       "   0.5874,\n",
       "   0.5874,\n",
       "   0.343,\n",
       "   0.26,\n",
       "   0.26,\n",
       "   0.3825,\n",
       "   0.5794,\n",
       "   0.5863],\n",
       "  u'sur_refl_b01[0][9]': [0.5883,\n",
       "   0.5591,\n",
       "   0.5263,\n",
       "   0.4325,\n",
       "   0.343,\n",
       "   0.4485,\n",
       "   0.626,\n",
       "   0.626,\n",
       "   0.5462,\n",
       "   0.3682,\n",
       "   0.3463]}}"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "simple_dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "## Create an RDD\n",
    "rdd_dataset = sc.parallelize(simple_dataset)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:475"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdd_dataset"
   ]
  },
  {
   "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
}