Precipitation_Analysis.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "(C) Copyright IBM Corp. 2016"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Set\n",
    "In this tutorial you will get acquainted with interactive notebooks by exploring and analyzing historical annual precipitation data.\n",
    "\n",
    "The raw precipitation data set is from [UNdata](http://data.un.org/), an Internet search engine for statistical databases provided by the United Nations Statistics Division. We will make use of a curated version of this data set in this tutorial. The measurements are in million cubic meter.\n",
    "\n",
    "1. Download the annual precipitation data in CSV format by clicking on this [link](https://cdsax.cloudant.com/public-samples/test/precipitation.csv).                 \n",
    "    **Note:** If you use Safari, then right click on the link and choose \"Download Linked File\". The CSV file will be downloaded to your \"Download\" folder.              \n",
    "1. Save the CSV file to your computer.\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Upload a File by Drag and Drop\n",
    "You can easily add the downloaded CSV file by dragging it from your computer and dropping it onto the drop off area of the **Data Source** panel on the **Palette**. \n",
    "Note that the file appears in the **Data Source** panel on the **Palette** and the file is saved in Object Storage.\n",
    "During the upload of the file, a progress bar indicates the status of the process. Below, you will learn how to access files in Object Storage.\n",
    "    \n",
    "<img src=\"https://cdsax.cloudant.com/public-samples/precipitation_analysis/precipitation_data_source.png\" width=\"20%\" height=\"20%\"> \n",
    "After adding a data source to your notebook, you can insert the credentials into your code for further access via the **Insert to code** function. If you want to delete files or containers, then use the ** Manage files** function. It takes you to the Object Storage dashboard where you have access to more options for organize your files."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Three important dependancies used in this notebook are:\n",
    "1. ** Spark-csv ** is a library that is massively used throughout the data science community to automatically infer the schema of your data.\n",
    "2. ** Scalaj-http **  is a fully featured http client for Scala \n",
    "3. **Spark-kernel-brunel** is a visualization framework which you are going to be using to visualize\n",
    "your data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 1: Import dependancies, jars and packages into our notebook.\n",
    "Click on the code cell below, then click the right arrow button (**&#9658;**) in the notebook **Toolbar** to run the code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "/* Update: As of [07.12.2016], the spark-kernel-brunel jar, spark-csv and scalaj-http dependencies \n",
    "are loaded by default in the Spark Bluemix Service */\n",
    "    //Import Dependancies\n",
    "    //%AddDeps com.databricks spark-csv_2.10 1.4.0 --transitive\n",
    "    //%AddDeps org.scalaj scalaj-http_2.10 1.1.4 --transitive\n",
    "    //Import Jar \n",
    "    //%AddJar -magic http://www.brunelvis.org/jar/spark-kernel-brunel-all-1.2.jar -f\n",
    "\n",
    "//Import Packages\n",
    "import scala.collection.breakOut\n",
    "import org.apache.spark.sql.SQLContext\n",
    "import org.apache.spark.rdd.PairRDDFunctions\n",
    "import org.apache.spark.sql.functions._\n",
    "import org.apache.spark.sql._"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can print your spark kernel configuration to ensure version compliance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Spark Context Version: 1.6.0\n",
      "Java version: 1.8.0\n",
      "Scala Release version: Some(2.10.5)\n",
      "_____________________\n",
      "Spark Context Config:\n",
      "(spark.eventLog.enabled,true)\n",
      "(spark.deploy.resourceScheduler.factory,org.apache.spark.deploy.master.EGOResourceSchedulerFactory)\n",
      "(spark.ui.retainedJobs,0)\n",
      "(spark.r.command,/usr/local/src/bluemix_jupyter_bundle.v8/R/bin/Rscript)\n",
      "(spark.shuffle.service.enabled,true)\n",
      "(spark.executor.extraJavaOptions,-Djava.security.egd=file:/dev/./urandom)\n",
      "(spark.executor.id,driver)\n",
      "(spark.port.maxRetries,512)\n",
      "(spark.sql.tungsten.enabled,false)\n",
      "(spark.repl.class.uri,http://10.143.133.52:40817)\n",
      "(spark.logConf,true)\n",
      "(spark.app.id,app-20160706132123-0270-037d6b55-d1db-4077-aa15-9e4c1612d83f)\n",
      "(spark.externalBlockStore.folderName,spark-ec312195-b75a-46d5-93e6-a8cdbfc28ab5)\n",
      "(spark.app.name,IBM Spark Kernel)\n",
      "(spark.ui.enabled,false)\n",
      "(spark.task.maxFailures,10)\n",
      "(spark.master,spark://yp-spark-dal09-env5-0028:7082)\n",
      "(spark.driver.port,58684)\n",
      "(spark.sql.unsafe.enabled,false)\n",
      "(spark.ui.retainedStages,0)\n",
      "(spark.history.fs.logDirectory,/gpfs/fs01/user/s275-101de6e70017ee-39207f492ff2/events)\n",
      "(spark.eventLog.dir,/gpfs/fs01/user/s275-101de6e70017ee-39207f492ff2/events)\n",
      "(spark.worker.ui.retainedExecutors,0)\n",
      "(spark.driver.host,10.143.133.52)\n",
      "(spark.shuffle.service.port,7340)\n",
      "(spark.master.rest.port,6070)\n"
     ]
    }
   ],
   "source": [
    "println(\"Spark Context Version: \" + sc.version);\n",
    "println(\"Java version: \" + scala.util.Properties.javaVersion);\n",
    "println(\"Scala Release version: \" + scala.util.Properties.releaseVersion);\n",
    "println(\"_____________________\");\n",
    "println(\"Spark Context Config:\");\n",
    "sc.getConf.getAll.foreach(println);\n",
    "println(\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 2: Access Object Storage\n",
    "\n",
    "Because the CSV file is located in Object Storage, we need a helper function to access the file using\n",
    "provided credentials. Run the cell below to define the method `setConfig()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "/*found at http://stackoverflow.com/questions/33725500/load-data-from-bluemix-object-store-in-spark*/\n",
    "\n",
    "def setConfig(name:String, dsConfiguration:String) : Unit = {\n",
    "    val pfx = \"fs.swift.service.\" + name\n",
    "    val settings:Map[String,String] = dsConfiguration.split(\"\\\\n\").\n",
    "        map(l=>(l.split(\":\",2)(0).trim(), l.split(\":\",2)(1).trim()))(breakOut)\n",
    "\n",
    "    val conf = sc.getConf\n",
    "    conf.set(pfx + \"auth.url\", settings.getOrElse(\"auth_url\",\"\"))\n",
    "    conf.set(pfx + \"tenant\", settings.getOrElse(\"tenantId\", \"\"))\n",
    "    conf.set(pfx + \"username\", settings.getOrElse(\"username\", \"\"))\n",
    "    conf.set(pfx + \"password\", settings.getOrElse(\"password\", \"\"))\n",
    "    conf.set(pfx + \"apikey\", settings.getOrElse(\"password\", \"\"))\n",
    "    conf.set(pfx + \"auth.endpoint.prefix\", \"endpoints\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 3: Insert data source credentials\n",
    "\n",
    "Click the cell below and select **Insert to code** of the precipitation.csv file in the **Data Source**\n",
    "Panel of the **Palette**. The neccesary credentials for accessing precipitation.csv will then be \n",
    "pasted into the code cell as a Scala dictionary. With the credentials, you can use the helper function\n",
    "to load it into a dataframe. **Note**: Adjust the variable name of the inserted Scala \n",
    "dictionary according to the code below and run the code cell."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 4: Load data into a `dataframe`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "done loading credentials\n",
      "created df\n"
     ]
    }
   ],
   "source": [
    "setConfig(\"spark\", credentials_3.toString())\n",
    "println(\"done loading credentials\")\n",
    "println(\"\")\n",
    "\n",
    "/*found at https://github.com/databricks/spark-csv*/\n",
    "val sqlctx = new SQLContext(sc);\n",
    "val scplain = sqlctx.sparkContext;\n",
    "sqlctx.setConf(\"spark.sql.shuffle.partitions\", \"10\");\n",
    "import sqlctx.implicits._\n",
    "\n",
    "/* Parenthesis allows line continuation which helps with readability */\n",
    "val df = (sqlctx.read\n",
    "    .format(\"com.databricks.spark.csv\")\n",
    "    .option(\"header\",\"true\")\n",
    "    .option(\"inferschema\",\"true\")\n",
    "    .option(\"mode\",\"DROPMALFORMED\")\n",
    "    .load(\"swift://notebooks.spark/precipitation.csv\")\n",
    ");\n",
    "\n",
    "println(\"created df\")\n",
    "println(\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Explore Data\n",
    "Now that we have the data in memory, we can explore and manipulate it.\n",
    "\n",
    "Print the first 5 rows of the data using the `show()` method.  Run the code cell below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Each row provides:\n",
    "\n",
    "* The Country or Area of the measurements\n",
    "* The annual precipitation for 1990, and for 1995 to 2009\n",
    "\n",
    "Using the `DataFrame` API, we can list all countries or areas, just run the cell below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array([Albania], [Algeria], [Andorra], [Anguilla], [Antigua and Barbuda], [Armenia], [Azerbaijan], [Bahrain], [Barbados], [Belarus], [Belgium], [Belize], [Benin], [Bermuda], [Bosnia and Herzegovina], [Botswana], [British Virgin Islands], [Brunei Darussalam], [Cameroon], [Central African Republic], [Chile], [China], [China, Hong Kong SAR], [China, Macao SAR], [Colombia], [Cote d'Ivoire], [Croatia], [Cuba], [Cyprus], [Czech Republic], [Denmark], [Dominican Republic], [Ecuador], [Egypt], [Estonia], [Finland], [France], [Gambia], [Georgia], [Germany], [Guinea], [Hungary], [India], [Iraq], [Israel], [Italy], [Jamaica], [Jordan], [Kazakhstan], [Kuwait], [Kyrgyzstan], [Latvia], [Lebanon], [Lithuania], [Luxembourg], [Madagascar], [Maldives], [Malta], [Ma..."
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select(\"Country or Area\").rdd.collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Plot\n",
    "When working with interactive notebooks, you can decide how to present results and information. So far, we have used normal print functions which are informative. We can also choose a visual way, using Brunel. Brunel takes data in a specific way. It requires a dataframe as an input and could take column values for x and y axes. You can preprocess the data to create a dataframe that holds the value to be visualized on the x and y axes."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The x axis will show the years so you can create an array holding the years."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "var years = df.columns.drop(1).map(r=>r.toString)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array(1990, 1995, 1996, 1997, 1998, 1999, 2000, 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009)"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "years"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The y axis will show the precipitation values for Germany so you can create an array to hold these."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var germanyDF = df.where($\"Country or Area\"===\"Germany\").drop(\"Country or Area\")\n",
    "var germany = germanyDF.collect()(0).toSeq.toArray.map(_.asInstanceOf[Double])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "germany"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We have two arrays that contain the year values and the precipitation amounts in Germany through these years. We index both arrays so we can later join them on these indexes. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val germanyIndexed = germany.zipWithIndex.map(_.swap)\n",
    "val yearsIndexed = years.zipWithIndex.map(_.swap)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The PairRDDFunctions class provide extra functions that can be performed on RDDs such as joins.\n",
    "Therefore we create a PairRDDFunction object. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val precipInGermany = new PairRDDFunctions(sc.parallelize(germanyIndexed))\n",
    "var yearsRDD = sc.parallelize(yearsIndexed)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We perform the joining on the indexes that we previously created and create a dataframe out of the new precipitation-year pair. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var precipByYearInGermany = precipInGermany.join(yearsRDD)\n",
    "var precipByYearInGermanyDF = precipByYearInGermany.map(row=>row._2).toDF(\"Precipitation\",\"Year\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "precipByYearInGermanyDF.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Brunel is invoked using the magic word **`%%brunel`**. Let us have a look at the annual precipitation of Germany."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%brunel\n",
    "data(\"precipByYearInGermanyDF\") bar x(Year) y(Precipitation)\n",
    "title(\"Precipitation in Germany\") axes(x:'Year',y:'Precipitation (million cubic meters)') interaction(none)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Raise Questions\n",
    "We explored and prepared the annual precipitation data. Now, we can raise more complex questions and make use of the data to answer them.\n",
    "## Question 1: Which countries or areas have the highest total precipitation?\n",
    "The first step we must take to answer this question is to compute the sums for annual precipitation. This can be done by specifying the columns to be summed and using the `+` sign."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "\n",
    "var totalPrecipByCountryDF=df.withColumn(\"Sum\",df(\"1990\") \n",
    "                                                                + df(\"1995\") + df(\"1996\")\n",
    "                                                                + df(\"1997\") + df(\"1998\")\n",
    "                                                                + df(\"1999\") + df(\"2000\")\n",
    "                                                                + df(\"2001\") + df(\"2002\")\n",
    "                                                                + df(\"2003\") + df(\"2004\") \n",
    "                                                                + df(\"2005\") + df(\"2006\") \n",
    "                                                                + df(\"2007\") + df(\"2008\")\n",
    "                                                                + df(\"2009\")).orderBy(desc(\"Sum\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we have an **additional** column called \"Sum\" in `totalPrecipByCountryDF`, containing the sums of the corresponding annual precipitation values. Because we are interested in the countries or areas with the highest precipitation, we sorted the `DataFrame` by total precipitation. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var precipitationByCountry = totalPrecipByCountryDF.select(totalPrecipByCountryDF(\"Country or Area\"),\n",
    "                                              totalPrecipByCountryDF(\"Sum\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can print the top 5 countries or areas with the highest total precipitation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "precipitationByCountry.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our next goal is to plot the top 5 countries in a bar chart together for better comparison."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var top5TotalPrecipByCountry = totalPrecipByCountryDF.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "top5TotalPrecipByCountry"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var top5TotalPrecipByCountryRDD= sc.parallelize(totalPrecipByCountryDF.take(5))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var top5TotalPrecipByCountryTransposed = top5TotalPrecipByCountryRDD.collect().toSeq.map(\n",
    "    r=>r.toSeq).transpose"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "top5TotalPrecipByCountryTransposed.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var top5PrecipitationCountries = sc.parallelize(top5TotalPrecipByCountryTransposed.map(r=>Row(r: _*)).drop(1))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "var top5PrecipByCountryIndexed = top5PrecipitationCountries.zipWithIndex.map(_.swap)\n",
    "var yearsIndexedWithRow = sc.parallelize(years).map(r => Row(r)).zipWithIndex.map(_.swap)\n",
    "var pairRdd  = new PairRDDFunctions(top5PrecipByCountryIndexed)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can join the top5PrecipByCountryIndexed and the yearsIndexedWithRow. Both objects\n",
    "have the same structure (index,Row(Field))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var top5PrecipByCountryWithYear = pairRdd.join(yearsIndexedWithRow)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The result is a PairRddFunction object of Tuples with key-value objects where the value is an Array of two Rows.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(14,([6200000.0,2545813.0,1293661.5,0.0,0.0],[2008]))\n"
     ]
    }
   ],
   "source": [
    "println(top5PrecipByCountryWithYear.take(2)(1))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can apply a map function to combine the year and the precipitation values rows and remove the indexing. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var top5PrecipByCountryWithYearArray = top5PrecipByCountryWithYear.map(r=>r._2).collect().map(\n",
    "                                                    p=> p._2.toSeq ++ p._1.toSeq).map(r=>Row(r:_*))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array([1998,0.0,2501223.0,1520000.0,0.0,951060.0])"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "top5PrecipByCountryWithYearArray.take(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now you can create a new dataframe holding the years and precipitation values for the top5 countries. The schema is defined below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import org.apache.spark.sql._\n",
    "import org.apache.spark.sql.types._\n",
    "\n",
    "val schema2 = StructType(\n",
    "    Array(\n",
    "    StructField(\"Year\",StringType,true),\n",
    "    StructField(\"China\",DoubleType,true),\n",
    "    StructField(\"Colombia\",DoubleType,true),\n",
    "    StructField(\"Venezuela\",DoubleType,true),\n",
    "    StructField(\"India\",DoubleType,true),\n",
    "    StructField(\"Chile\",DoubleType,true)\n",
    "))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var precipByCountryWithYearDF = sqlctx.createDataFrame(sc.parallelize(top5PrecipByCountryWithYearArray),schema2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+---------+---------+-----------+---------+-----------+\n",
      "|Year|    China| Colombia|  Venezuela|    India|      Chile|\n",
      "+----+---------+---------+-----------+---------+-----------+\n",
      "|1998|      0.0|2501223.0|  1520000.0|      0.0|   951060.0|\n",
      "|2008|6200000.0|2545813.0|  1293661.5|      0.0|        0.0|\n",
      "|1990|      0.0|2190672.0|  1630000.0|      0.0|  1152225.0|\n",
      "|2000|6009200.0|2529872.0|  1360000.0|      0.0|  1160289.0|\n",
      "|2002|6261000.0|2044818.0|  1420000.0|      0.0|  1306221.0|\n",
      "|2006|5784000.0|2509840.0|  1060000.0|4000000.0|  1181614.0|\n",
      "|2004|5687600.0|2154068.0|  1520000.0|4000000.0|1272096.625|\n",
      "|1996|      0.0|2599285.0|  1420000.0|      0.0|   897224.0|\n",
      "|2007|5776300.0|2492041.0|1335846.125|4000000.0|800225.1875|\n",
      "|2009|5596600.0|      0.0|   936000.0|      0.0|        0.0|\n",
      "|2005|6101000.0|2467871.0|  1560000.0|4000000.0| 1231319.25|\n",
      "|1995|      0.0|2399888.0|  1520000.0|      0.0|   971392.0|\n",
      "|1997|      0.0|1945543.0|  1470000.0|      0.0|  1315783.0|\n",
      "|2001|5812200.0|2059180.0|  1360000.0|      0.0|  1239664.0|\n",
      "|2003|6041600.0|2263478.0|  1195231.0|      0.0|  1377889.0|\n",
      "|1999|      0.0|2897358.0|  1560000.0|      0.0|  1083755.0|\n",
      "+----+---------+---------+-----------+---------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "precipByCountryWithYearDF.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Brunel supports multi-layering so you can take advantage of it in the following way. It also supports interactivity. Hover over the points on the graph to get the precipitation value for the corresponding year. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "         <link rel=\"stylesheet\" type=\"text/css\" href=\"https://brunelvis.org/js/brunel.1.2.css\" charset=\"utf-8\">\n",
       "         <link rel=\"stylesheet\" type=\"text/css\" href=\"https://brunelvis.org/js/sumoselect.css\" charset=\"utf-8\">\n",
       "         <style>  </style>\n",
       "         <div id=\"controlsId7d8c0dab-e7af-4e17-9728-3e6b898d3607\" class=\"brunel\"/>\n",
       "<svg id=\"visidd4cc5257-2607-4c50-b4a7-9d9fc32bd0b4\" width=\"500\" height=\"400\"></svg>\n",
       "\n",
       "<script>\n",
       "require.config({\n",
       "            waitSeconds: 60,\n",
       "            paths: {\n",
       "                'd3': '//cdnjs.cloudflare.com/ajax/libs/d3/3.5.5/d3.min',\n",
       "                'topojson' : '//cdnjs.cloudflare.com/ajax/libs/topojson/1.6.20/topojson.min',\n",
       "                'brunel' : 'https://brunelvis.org/js/brunel.1.2.min',\n",
       "                'brunelControls' : 'https://brunelvis.org/js/brunel.controls.1.2.min'\n",
       "            },\n",
       "\n",
       "            shim: {\n",
       "               'brunel' : {\n",
       "                    exports: 'BrunelD3',\n",
       "                    deps: ['d3', 'topojson'],\n",
       "                    init: function() {\n",
       "                       return {\n",
       "                         BrunelD3 : BrunelD3,\n",
       "                         BrunelData : BrunelData\n",
       "                      }\n",
       "                    }\n",
       "                },\n",
       "               'brunelControls' : {\n",
       "                    exports: 'BrunelEventHandlers',\n",
       "                    init: function() {\n",
       "                       return {\n",
       "                         BrunelEventHandlers: BrunelEventHandlers,\n",
       "                         BrunelJQueryControlFactory: BrunelJQueryControlFactory\n",
       "                      }\n",
       "                    }\n",
       "                }\n",
       "\n",
       "            }\n",
       "\n",
       "        });\n",
       "\n",
       "        require([\"d3\"], function(d3) {\n",
       "        require([\"brunel\", \"brunelControls\"], function(brunel, brunelControls) {\n",
       "\n",
       "            function  BrunelVis(visId) {\n",
       "  \"use strict\"; // Strict Mode\n",
       "  var datasets = [],                               // Array of datasets for the original data\n",
       "      pre = function(d, i) { return d },           // Default pre-process does nothing\n",
       "      post = function(d, i) { return d },          // Default post-process does nothing\n",
       "      transitionTime = 200,                        // Transition time for animations\n",
       "      charts = [],                                 // The charts in the system\n",
       "      vis = d3.select('#' + visId).attr('class', 'brunel');  // the SVG container\n",
       "\n",
       "  // Define chart #1 in the visualization //////////////////////////////////////////////////////////\n",
       "\n",
       "  charts[0] = function(parentNode, filterRows) {\n",
       "    var geom = BrunelD3.geometry(parentNode || vis.node(), 0, 0, 1, 1, 32, 84, 53, 96),\n",
       "      elements = [];                               // Array of elements in this chart\n",
       "\n",
       "    // Define groups for the chart parts ///////////////////////////////////////////////////////////\n",
       "\n",
       "    var chart = vis.append('g').attr('class', 'chart1')\n",
       "      .attr('transform','translate(' + geom.chart_left + ',' + geom.chart_top + ')');\n",
       "    chart.append('rect').attr('class', 'background').attr('width', geom.chart_right-geom.chart_left).attr('height', geom.chart_bottom-geom.chart_top);\n",
       "    var interior = chart.append('g').attr('class', 'interior')\n",
       "      .attr('transform','translate(' + geom.inner_left + ',' + geom.inner_top + ')')\n",
       "      .attr('clip-path', 'url(#clip_visidd4cc5257-2607-4c50-b4a7-9d9fc32bd0b4_1)');\n",
       "    interior.append('rect').attr('class', 'inner').attr('width', geom.inner_width).attr('height', geom.inner_height);\n",
       "    var overlay = interior.append('g').attr('class', 'element')\n",
       "      .attr('class', 'overlay').style('cursor','move').style('fill','none').style('pointer-events','all');\n",
       "    var zoom = d3.behavior.zoom().on('zoom', function() {build(-1)} );\n",
       "    overlay.append('rect').attr('class', 'overlay')\n",
       "      .attr('width', geom.inner_width)\n",
       "      .attr('height', geom.inner_height)\n",
       "      .call(zoom);\n",
       "    var axes = chart.append('g').attr('class', 'axis')\n",
       "      .attr('transform','translate(' + geom.inner_left + ',' + geom.inner_top + ')');\n",
       "    var legends = chart.append('g').attr('class', 'legend')\n",
       "      .attr('transform','translate(' + (geom.chart_right-geom.chart_left - 3) + ',' + 0 + ')');\n",
       "    vis.append('clipPath').attr('id', 'clip_visidd4cc5257-2607-4c50-b4a7-9d9fc32bd0b4_1').append('rect')\n",
       "      .attr('x', -1).attr('y', -1)\n",
       "      .attr('width', geom.inner_width+2).attr('height', geom.inner_height+2);\n",
       "    chart.append('text').attr('class', 'title header').text('Top 5 Countries with highest precipitation').style('text-anchor', 'middle').attr('dy','0.8em')\n",
       "      .attr('x','50%')\n",
       "      .attr('y',0);\n",
       "\n",
       "    // Scales //////////////////////////////////////////////////////////////////////////////////////\n",
       "\n",
       "    var scale_x = d3.scale.ordinal()\n",
       "      .domain(['1990', '1995', '1996', '1997', '1998', '1999', '2000', '2001', '2002', '2003', '2004', \n",
       "        '2005', '2006', '2007', '2008', '2009'])\n",
       "      .rangePoints([0, geom.inner_width], 1);\n",
       "    var scale_inner = d3.scale.linear().domain([0,1])\n",
       "      .range([-0.5, 0.5]);\n",
       "    var scale_y = d3.scale.sqrt()\n",
       "      .domain([-1000000, 7000000])\n",
       "      .range([geom.inner_height,0]);\n",
       "    zoom.y(scale_y);\n",
       "\n",
       "    // Axes ////////////////////////////////////////////////////////////////////////////////////////\n",
       "\n",
       "    axes.append('g').attr('class', 'x axis')\n",
       "      .attr('transform','translate(0,' + geom.inner_height + ')');\n",
       "    axes.select('g.axis.x').append('text').attr('class', 'title')\n",
       "      .attr('text-anchor', 'middle')\n",
       "      .attr('x', geom.inner_width/2)\n",
       "      .attr('y', geom.inner_bottom - 6.0)\n",
       "      .text('Years');\n",
       "    axes.append('g').attr('class', 'y axis')\n",
       "      .attr('transform','translate(geom.chart_left, 0)');\n",
       "    axes.select('g.axis.y').append('text').attr('class', 'title')\n",
       "      .attr('text-anchor', 'middle')\n",
       "      .attr('x', -geom.inner_height/2)\n",
       "      .attr('y', 6-geom.inner_left).attr('dy', '0.7em').attr('transform', 'rotate(270)')\n",
       "      .text('Precipitation (million cubic meters)');\n",
       "\n",
       "    var axis_bottom = d3.svg.axis()\n",
       "      .scale(scale_x).innerTickSize(3).outerTickSize(0)\n",
       "      .tickValues(['1990', '1996', '1998', '2000', '2002', '2004', '2006', '2008', \n",
       "        '1990', '1996', '1998', '2000', '2002', '2004', '2006', '2008'])\n",
       "      .ticks(Math.min(10, Math.round(geom.inner_width / 39.0)));\n",
       "    var axis_left = d3.svg.axis().orient('left')\n",
       "      .scale(scale_y).innerTickSize(3).outerTickSize(0)\n",
       "      .tickFormat(  function(x) { return BrunelData.Data.format(x/1e6) + 'M' });\n",
       "\n",
       "    function buildAxes() {\n",
       "      axes.select('g.axis.x').call(axis_bottom).selectAll('.tick text')\n",
       "        .style('text-anchor', 'end')\n",
       "        .attr('dx', '-.3em')\n",
       "        .attr('dy', '.6em')\n",
       "        .attr('transform', function(d) { return 'rotate(-45)' });\n",
       "      axes.select('g.axis.y').call(axis_left);\n",
       "    }\n",
       "\n",
       "    // Define element #1 ///////////////////////////////////////////////////////////////////////////\n",
       "\n",
       "    elements[0] = function() {\n",
       "      var original, processed,           // data sets passed in and then transformed\n",
       "        element, data,                   // Brunel element information and brunel data\n",
       "        selection;                       // D3 selection\n",
       "      var elementGroup = interior.append('g').attr('class', 'element1'),\n",
       "        main = elementGroup.append('g').attr('class', 'main'),\n",
       "        labels = BrunelD3.undoTransform(elementGroup.append('g').attr('class', 'labels'), elementGroup);\n",
       "\n",
       "      function makeData() {\n",
       "        original = datasets[0];\n",
       "        if (filterRows) original = original.retainRows(filterRows);\n",
       "        processed = pre(original, 0)\n",
       "          .reduce('Year; Colombia; China; Venezuela; India; Chile; #series; #values; #row; #count')\n",
       "          .series('Colombia,China,Venezuela,India,Chile;Year');\n",
       "        processed = post(processed, 0);\n",
       "        var f0 = processed.field('#series'),\n",
       "          f1 = processed.field('#values'),\n",
       "          f2 = processed.field('Year'),\n",
       "          f3 = processed.field('#row');\n",
       "        var keyFunc = function(d) { return f3.value(d)+ '|' + f0.value(d) };\n",
       "        data = {\n",
       "          $series:      function(d) { return f0.value(d.row) },\n",
       "          $values:      function(d) { return f1.value(d.row) },\n",
       "          Year:         function(d) { return f2.value(d.row) },\n",
       "          $row:         function(d) { return f3.value(d.row) },\n",
       "          $series_f:    function(d) { return f0.valueFormatted(d.row) },\n",
       "          $values_f:    function(d) { return f1.valueFormatted(d.row) },\n",
       "          Year_f:       function(d) { return f2.valueFormatted(d.row) },\n",
       "          $row_f:       function(d) { return f3.valueFormatted(d.row) },\n",
       "          _split:       function(d) { return f0.value(d.row) },\n",
       "          _key:         keyFunc,\n",
       "          _rows:        BrunelD3.makeRowsWithKeys(keyFunc, processed.rowCount())\n",
       "        };\n",
       "      }\n",
       "      // Aesthetic Functions\n",
       "      var scale_color = d3.scale.ordinal()\n",
       "        .domain(['Colombia', 'China', 'Venezuela', 'India', 'Chile'])\n",
       "        .range([ '#00538A', '#C10020', '#F4C800', '#007D34', '#803E75', '#FF6800', \n",
       "          '#817066', '#FFB300', '#F6768E', '#93AA00', '#53377A', '#FF8E00', '#B32851', \n",
       "          '#CEA262', '#FF7A5C', '#7F180D', '#593315', '#F13A13', '#232C16']);\n",
       "      var color = function(d) { return scale_color(data.$series(d)) };\n",
       "      BrunelD3.addLegend(legends, 'Series', scale_color, scale_color.domain());\n",
       "\n",
       "      // Build element from data ///////////////////////////////////////////////////////////////////\n",
       "\n",
       "      function build(transitionMillis) {\n",
       "        element = elements[0];\n",
       "        var x = function(d) { return scale_x(data.Year(d))};\n",
       "        var w = 0.9 * geom.inner_width/16;\n",
       "        var y = function(d) { return scale_y(data.$values(d))};\n",
       "        var h = geom.default_point_size;\n",
       "        selection = main.selectAll('*').data(data._rows, function(d) { return d.key});\n",
       "        selection.enter().append('circle').attr('class',  'element point filled');\n",
       "\n",
       "        BrunelD3.trans(selection,transitionMillis)\n",
       "          .attr('cx',function(d) { return scale_x(data.Year(d))})\n",
       "          .attr('cy',function(d) { return scale_y(data.$values(d))})\n",
       "          .attr('r',Math.min(0.9 * geom.inner_width/16, geom.default_point_size) / 2)\n",
       "          .style('fill', color);\n",
       "\n",
       "        var tooltipLabeling = {\n",
       "          method: 'top', \n",
       "          fit: true, granularity: 0,\n",
       "          content: function(d) {\n",
       "            return d.row == null ? null : 'Value for '\n",
       "\t\t\t+ '<span class=\"field\">' + data.$series_f(d) + '</span>'\n",
       "\t\t\t+ ' is '\n",
       "\t\t\t+ '<span class=\"field\">' + data.$values_f(d) + '</span>'\n",
       "          }\n",
       "        };\n",
       "        BrunelD3.addTooltip(selection, tooltipLabeling, geom);\n",
       "\n",
       "        BrunelD3.trans(selection.exit(),transitionMillis/3)\n",
       "          .style('opacity', 0.5).each( function() {this.remove(); if (this.__label__) this.__label__.remove()});\n",
       "      }\n",
       "\n",
       "      return {\n",
       "        data:           function() { return processed },\n",
       "        internal:       function() { return data },\n",
       "        selection:      function() { return selection },\n",
       "        makeData:       makeData,\n",
       "        build:          build,\n",
       "        fields: {\n",
       "          x:            ['Year'],\n",
       "          y:            ['Colombia', 'China', 'Venezuela', 'India', 'Chile'],\n",
       "          key:          ['#row', '#series'],\n",
       "          color:        ['#series']\n",
       "        }\n",
       "      };\n",
       "    }();\n",
       "\n",
       "    // Define element #2 ///////////////////////////////////////////////////////////////////////////\n",
       "\n",
       "    elements[1] = function() {\n",
       "      var original, processed,           // data sets passed in and then transformed\n",
       "        element, data,                   // Brunel element information and brunel data\n",
       "        selection;                       // D3 selection\n",
       "      var elementGroup = interior.append('g').attr('class', 'element2'),\n",
       "        main = elementGroup.append('g').attr('class', 'main'),\n",
       "        labels = BrunelD3.undoTransform(elementGroup.append('g').attr('class', 'labels'), elementGroup);\n",
       "\n",
       "      function makeData() {\n",
       "        original = datasets[0];\n",
       "        if (filterRows) original = original.retainRows(filterRows);\n",
       "        processed = pre(original, 0)\n",
       "          .reduce('Year; Colombia; China; Venezuela; India; Chile; #series; #row; #count')\n",
       "          .series('Colombia,China,Venezuela,India,Chile;Year')\n",
       "          .sortRows('Year:ascending');\n",
       "        processed = post(processed, 0);\n",
       "        var f0 = processed.field('#series'),\n",
       "          f1 = processed.field('#values'),\n",
       "          f2 = processed.field('Year'),\n",
       "          f3 = processed.field('#row');\n",
       "        var keyFunc = function(d) { return f0.value(d) };\n",
       "        data = {\n",
       "          $series:      function(d) { return f0.value(d.row) },\n",
       "          $values:      function(d) { return f1.value(d.row) },\n",
       "          Year:         function(d) { return f2.value(d.row) },\n",
       "          $row:         function(d) { return f3.value(d.row) },\n",
       "          $series_f:    function(d) { return f0.valueFormatted(d.row) },\n",
       "          $values_f:    function(d) { return f1.valueFormatted(d.row) },\n",
       "          Year_f:       function(d) { return f2.valueFormatted(d.row) },\n",
       "          $row_f:       function(d) { return f3.valueFormatted(d.row) },\n",
       "          _split:       function(d) { return f0.value(d.row) },\n",
       "          _key:         keyFunc,\n",
       "          _rows:        BrunelD3.makeRowsWithKeys(keyFunc, processed.rowCount())\n",
       "        };\n",
       "      }\n",
       "      // Aesthetic Functions\n",
       "      var scale_color = d3.scale.ordinal()\n",
       "        .domain(['Colombia', 'China', 'Venezuela', 'India', 'Chile'])\n",
       "        .range([ '#00538A', '#C10020', '#F4C800', '#007D34', '#803E75', '#FF6800', \n",
       "          '#817066', '#FFB300', '#F6768E', '#93AA00', '#53377A', '#FF8E00', '#B32851', \n",
       "          '#CEA262', '#FF7A5C', '#7F180D', '#593315', '#F13A13', '#232C16']);\n",
       "      var color = function(d) { return scale_color(data.$series(d)) };\n",
       "      BrunelD3.addLegend(legends, 'Series', scale_color, scale_color.domain());\n",
       "\n",
       "      // Build element from data ///////////////////////////////////////////////////////////////////\n",
       "\n",
       "      function build(transitionMillis) {\n",
       "        element = elements[1];\n",
       "        var x = function(d) { return scale_x(data.Year(d))};\n",
       "        var w = 0.9 * geom.inner_width/16;\n",
       "        var y = function(d) { return scale_y(data.$values(d))};\n",
       "        var h = geom.default_point_size;\n",
       "        var path = d3.svg.line().x(x).y(y);\n",
       "        var splits = BrunelD3.makePathSplits(data, path, x);\n",
       "        selection = main.selectAll('*').data(splits, function(d) { return d.key});\n",
       "        selection.enter().append('path').attr('class',  'element line');\n",
       "\n",
       "        BrunelD3.trans(selection,transitionMillis)\n",
       "          .attr('d', function(d) { return d.path })\n",
       "          .style('stroke', color);\n",
       "\n",
       "        BrunelD3.trans(selection.exit(),transitionMillis/3)\n",
       "          .style('opacity', 0.5).each( function() {this.remove(); if (this.__label__) this.__label__.remove()});\n",
       "      }\n",
       "\n",
       "      return {\n",
       "        data:           function() { return processed },\n",
       "        internal:       function() { return data },\n",
       "        selection:      function() { return selection },\n",
       "        makeData:       makeData,\n",
       "        build:          build,\n",
       "        fields: {\n",
       "          x:            ['Year'],\n",
       "          y:            ['Colombia', 'China', 'Venezuela', 'India', 'Chile'],\n",
       "          key:          ['#series'],\n",
       "          color:        ['#series']\n",
       "        }\n",
       "      };\n",
       "    }();\n",
       "\n",
       "    function build(time) {\n",
       "      var first = elements[0].data() == null;\n",
       "      if (first) time = 0; // No transition for first call\n",
       "      buildAxes();\n",
       "      if (first || time > -1) {\n",
       "        elements[0].makeData();\n",
       "        elements[1].makeData();\n",
       "      }\n",
       "      elements[0].build(time);\n",
       "      elements[1].build(time);\n",
       "    }\n",
       "\n",
       "    // Expose the following components of the chart\n",
       "    return {build : build, elements : elements, scales: {x:scale_x, y:scale_y}};\n",
       "    }();\n",
       "\n",
       "  function setData(rowData, i) { datasets[i||0] = BrunelD3.makeData(rowData) }\n",
       "  function updateAll(time) { charts.forEach(function(x) {x.build(time || 0)}) }\n",
       "  function buildAll() {\n",
       "    for (var i=0;i<arguments.length;i++) setData(arguments[i], i);\n",
       "    updateAll(transitionTime);\n",
       "  }\n",
       "\n",
       "  return {\n",
       "    dataPreProcess:     function(f) { if (f) pre = f; return pre },\n",
       "    dataPostProcess:    function(f) { if (f) post = f; return post },\n",
       "    data:               function(d,i) { if (d) setData(d,i); return datasets[i||0] },\n",
       "    visId:              visId,\n",
       "    build:              buildAll,\n",
       "    rebuild:            updateAll,\n",
       "    charts:             charts\n",
       "  }\n",
       "}\n",
       "\n",
       "// Data Tables /////////////////////////////////////////////////////////////////////////////////////\n",
       "\n",
       "var table1 = {\n",
       "   names: ['Year', 'Colombia', 'China', 'Venezuela', 'India', 'Chile'], \n",
       "   options: ['string', 'numeric', 'numeric', 'numeric', 'numeric', 'numeric'], \n",
       "   rows: [['1998', 2501223, 0, 1520000, 0, 951060], ['2008', 2545813, 6200000, 1293661.5, 0, 0],\n",
       "  ['1990', 2190672, 0, 1630000, 0, 1152225], ['2000', 2529872, 6009200, 1360000, 0, 1160289],\n",
       "  ['2002', 2044818, 6261000, 1420000, 0, 1306221],\n",
       "  ['2006', 2509840, 5784000, 1060000, 4000000, 1181614],\n",
       "  ['2004', 2154068, 5687600, 1520000, 4000000, 1272096.625],\n",
       "  ['1996', 2599285, 0, 1420000, 0, 897224],\n",
       "  ['2007', 2492041, 5776300, 1335846.125, 4000000, 800225.1875], ['2009', 0, 5596600, 936000, 0, 0],\n",
       "  ['2005', 2467871, 6101000, 1560000, 4000000, 1231319.25], ['1995', 2399888, 0, 1520000, 0, 971392],\n",
       "  ['1997', 1945543, 0, 1470000, 0, 1315783], ['2001', 2059180, 5812200, 1360000, 0, 1239664],\n",
       "  ['2003', 2263478, 6041600, 1195231, 0, 1377889], ['1999', 2897358, 0, 1560000, 0, 1083755]]\n",
       "};\n",
       "\n",
       "// Call Code to Build the system ///////////////////////////////////////////////////////////////////\n",
       "\n",
       "var v = new BrunelVis('visidd4cc5257-2607-4c50-b4a7-9d9fc32bd0b4');\n",
       "v.build(table1);\n",
       "\n",
       "            \"\"\n",
       "        });\n",
       "        });\n",
       "        </script>"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%brunel\n",
    "data(\"precipByCountryWithYearDF\") title(\"Top 5 Countries with highest precipitation\") \n",
    "x(Year) y(Colombia,China,Venezuela,India,Chile) \n",
    "color(#series) axes(x:'Years',y:'Precipitation (million cubic meters)')\n",
    "tooltip('Value for ',#series, ' is ',#values) \n",
    "+ line x(Year) y(Colombia,China,Venezuela,India,Chile) color(#series) \n",
    "\n",
    "//Resolution to https://github.com/Brunel-Visualization/Brunel/issues/104 should address the problem\n",
    "//with the y-axis starting below 0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "Now compare the annual precipitation values of the top 5 countries. China has the highest annual precipitation, followed by Colombia. It is also obvious that some values are missing, which makes comparison difficult. Now, we want to know the percentage values of the precipitation for the top 5 countries in relation to the rest. A pie chart is ideal to show this kind of information."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var top5 = precipitationByCountry.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var other = precipitationByCountry.collect().drop(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "other"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var totalPrecpitationForOther:Double =  0.0\n",
    "\n",
    "for(x <- other)\n",
    "{\n",
    "    totalPrecpitationForOther += x(1).asInstanceOf[Double]\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "totalPrecpitationForOther"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var top5WithOther = top5 :+ Row(\"Other\", totalPrecpitationForOther)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import org.apache.spark.sql._\n",
    "import org.apache.spark.sql.types._\n",
    "\n",
    "val schema3 = StructType(\n",
    "    Array(\n",
    "    StructField(\"CountryOrArea\",StringType,true),\n",
    "    StructField(\"Precipitation\", DoubleType,true)\n",
    "))\n",
    "\n",
    "var top5WithOtherDF = sqlctx.createDataFrame(sc.parallelize(top5WithOther),schema3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "top5WithOtherDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%brunel\n",
    "stack polar data(\"top5WithOtherDF\") y(Precipitation) title(\"Annual precipitation percentage\") \n",
    "color(CountryOrArea) label(Precipitation) percent(Precipitation)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "In this pie chart one can recognize that from the 91 countries and areas we consider, nearly a quarter of the precipitation for the observed years was in China, and over half of the precipitation was in the top 5 countries."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Question 2: Which countries show a negative trend in annual precipitation ?\n",
    "Going through each row of a DataFrame and looking at the numbers is not a viable solution to determine trends. Plotting a bar chart for each country or area is possible, but inconvenient. One possiblity to determine trends in annual precipitation is to fit a line to the data points. Run the code cells below to see the bar chart and the trend for annual precipitation for Chile.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var chile = df.filter(df(\"Country or Area\") === \"Chile\").drop(\"Country or Area\")\n",
    "var chileValues = chile.collect()(0).toSeq.toArray.map(_.asInstanceOf[Double])\n",
    "\n",
    "//var chilePrecipitationValues = chileValues.map(_.toDouble)\n",
    "val chileIndexed = chileValues.zipWithIndex.map(_.swap)\n",
    "val chileIndexedPairRDD = new PairRDDFunctions(sc.parallelize(chileIndexed))\n",
    "var precipByYearChile = chileIndexedPairRDD.join(sc.parallelize(yearsIndexed))\n",
    "var precipByYearChileDF = precipByYearChile.map(row=>row._2).toDF(\"Precipitation\",\"Year\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "precipByYearChileDF.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%brunel\n",
    "data(\"precipByYearChileDF\") bar x(Year) y(Precipitation) title(\"Precipitation in Chile\") \n",
    "axes(x:'Years',y:'Precipitation (million cubic meters)') interaction(none)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "precipByYearChileDF.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Exploring the data above, one can spot that the precipitation values for 2008 and 2009 are zero. You can filter \n",
    "these out by utilizing the filter function. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var yearsChile = years.filter(x=> x.toInt<2008)\n",
    "chileValues = chileValues.filter(r=>r>0.0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In order to fit a line through the data points, we need to compute the slope and the y-intercept. Therefore, \n",
    "we define three new functions. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "/* This method computes the slope of a fitted line for given data points. */\n",
    "def computeSlope( a:Array[Double], b:Array[Double]) : Double = {\n",
    "     \n",
    "     val xmean = a.sum / a.length\n",
    "     val ymean = b.sum / b.length\n",
    "     \n",
    "     val xdelta = a.map(r=> r - xmean)\n",
    "     val ydelta = b.map(w=> w - ymean)\n",
    "     \n",
    "     val productXY = (xdelta,ydelta).zipped.map((x,y) => x*y)\n",
    "     val productXX = xdelta.map(x => x*x)\n",
    "     \n",
    "     return productXY.sum/productXX.sum\n",
    "}\n",
    "\n",
    "/* This method computes the y-intercept */\n",
    "def computeYIntercept(a:Array[Double], b:Array[Double],slope:Double) : Double = {\n",
    "    \n",
    "    val xmean = a.sum / a.length\n",
    "    val ymean = b.sum / b.length\n",
    "     \n",
    "    return ymean - (slope*xmean)\n",
    "}\n",
    "\n",
    "def bestFit(a:Array[Double], slope: Double, yIntercept:Double):Array[Double]  = {\n",
    "  \n",
    "  return a.map(w=>  (w * slope) + yIntercept)\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To determine whether the trend is positive or negative, we only have to look at the slope. Of course, we had to exclude any datapoints correponding to value of 0.0."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var slope = computeSlope(yearsChile.map(_.toDouble),chileValues.map(_.toDouble))\n",
    "println(\"Slope: \" + slope)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var yIntercept = computeYIntercept(years.map(_.toDouble),chileValues.map(_.toDouble),slope)\n",
    "println(\"y-Intercept: \" + yIntercept)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var bestFitLine = bestFit(years.map(_.toDouble), slope,yIntercept)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We computed the best fit line so we are going to create a dataframe with three fields: precipitation, year, and \n",
    "the best fit line values. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var bestFitLineIndexed = bestFitLine.zipWithIndex.map(_.swap)\n",
    "var precipByYearChileBestLine = precipByYearChile.join(sc.parallelize(bestFitLineIndexed)).filter(r=>r._2._1._1>0)\n",
    "var precip = precipByYearChileBestLine.values.map(r=>Row(r._1._1,r._1._2,r._2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "val schema4 = StructType(\n",
    "    Array(\n",
    "    StructField(\"Precipitation\",DoubleType,true),\n",
    "    StructField(\"Year\",StringType,true),\n",
    "    StructField(\"BestLine\",DoubleType,true)\n",
    "))\n",
    "\n",
    "var precipBestLineDF = sqlctx.createDataFrame(precip,schema4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+----+------------------+\n",
      "|Precipitation|Year|          BestLine|\n",
      "+-------------+----+------------------+\n",
      "|     951060.0|1998|1120906.7870451212|\n",
      "|    1152225.0|1990| 1077291.482628718|\n",
      "|    1160289.0|2000| 1131810.613149222|\n",
      "|    1306221.0|2002|1142714.4392533228|\n",
      "|    1181614.0|2006|1164522.0914615244|\n",
      "|  1272096.625|2004|1153618.2653574236|\n",
      "|     897224.0|1996|1110002.9609410204|\n",
      "|  800225.1875|2007|1169974.0045135748|\n",
      "|   1231319.25|2005| 1159070.178409474|\n",
      "|     971392.0|1995|  1104551.04788897|\n",
      "|    1315783.0|1997|1115454.8739930708|\n",
      "|    1239664.0|2001|1137262.5262012724|\n",
      "|    1377889.0|2003|1148166.3523053732|\n",
      "|    1083755.0|1999|1126358.7000971716|\n",
      "+-------------+----+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "precipBestLineDF.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "         <link rel=\"stylesheet\" type=\"text/css\" href=\"https://brunelvis.org/js/brunel.1.2.css\" charset=\"utf-8\">\n",
       "         <link rel=\"stylesheet\" type=\"text/css\" href=\"https://brunelvis.org/js/sumoselect.css\" charset=\"utf-8\">\n",
       "         <style>  </style>\n",
       "         <div id=\"controlsId9e8e5b17-bafb-4304-8ae7-72e63f13d488\" class=\"brunel\"/>\n",
       "<svg id=\"visidd600d1d0-9993-4f7c-b9b2-4f2380ec32f2\" width=\"500\" height=\"400\"></svg>\n",
       "\n",
       "<script>\n",
       "require.config({\n",
       "            waitSeconds: 60,\n",
       "            paths: {\n",
       "                'd3': '//cdnjs.cloudflare.com/ajax/libs/d3/3.5.5/d3.min',\n",
       "                'topojson' : '//cdnjs.cloudflare.com/ajax/libs/topojson/1.6.20/topojson.min',\n",
       "                'brunel' : 'https://brunelvis.org/js/brunel.1.2.min',\n",
       "                'brunelControls' : 'https://brunelvis.org/js/brunel.controls.1.2.min'\n",
       "            },\n",
       "\n",
       "            shim: {\n",
       "               'brunel' : {\n",
       "                    exports: 'BrunelD3',\n",
       "                    deps: ['d3', 'topojson'],\n",
       "                    init: function() {\n",
       "                       return {\n",
       "                         BrunelD3 : BrunelD3,\n",
       "                         BrunelData : BrunelData\n",
       "                      }\n",
       "                    }\n",
       "                },\n",
       "               'brunelControls' : {\n",
       "                    exports: 'BrunelEventHandlers',\n",
       "                    init: function() {\n",
       "                       return {\n",
       "                         BrunelEventHandlers: BrunelEventHandlers,\n",
       "                         BrunelJQueryControlFactory: BrunelJQueryControlFactory\n",
       "                      }\n",
       "                    }\n",
       "                }\n",
       "\n",
       "            }\n",
       "\n",
       "        });\n",
       "\n",
       "        require([\"d3\"], function(d3) {\n",
       "        require([\"brunel\", \"brunelControls\"], function(brunel, brunelControls) {\n",
       "\n",
       "            function  BrunelVis(visId) {\n",
       "  \"use strict\"; // Strict Mode\n",
       "  var datasets = [],                               // Array of datasets for the original data\n",
       "      pre = function(d, i) { return d },           // Default pre-process does nothing\n",
       "      post = function(d, i) { return d },          // Default post-process does nothing\n",
       "      transitionTime = 200,                        // Transition time for animations\n",
       "      charts = [],                                 // The charts in the system\n",
       "      vis = d3.select('#' + visId).attr('class', 'brunel');  // the SVG container\n",
       "\n",
       "  // Define chart #1 in the visualization //////////////////////////////////////////////////////////\n",
       "\n",
       "  charts[0] = function(parentNode, filterRows) {\n",
       "    var geom = BrunelD3.geometry(parentNode || vis.node(), 0, 0, 1, 1, 32, 84, 53, 3),\n",
       "      elements = [];                               // Array of elements in this chart\n",
       "\n",
       "    // Define groups for the chart parts ///////////////////////////////////////////////////////////\n",
       "\n",
       "    var chart = vis.append('g').attr('class', 'chart1')\n",
       "      .attr('transform','translate(' + geom.chart_left + ',' + geom.chart_top + ')');\n",
       "    chart.append('rect').attr('class', 'background').attr('width', geom.chart_right-geom.chart_left).attr('height', geom.chart_bottom-geom.chart_top);\n",
       "    var interior = chart.append('g').attr('class', 'interior')\n",
       "      .attr('transform','translate(' + geom.inner_left + ',' + geom.inner_top + ')')\n",
       "      .attr('clip-path', 'url(#clip_visidd600d1d0-9993-4f7c-b9b2-4f2380ec32f2_1)');\n",
       "    interior.append('rect').attr('class', 'inner').attr('width', geom.inner_width).attr('height', geom.inner_height);\n",
       "    var axes = chart.append('g').attr('class', 'axis')\n",
       "      .attr('transform','translate(' + geom.inner_left + ',' + geom.inner_top + ')');\n",
       "    vis.append('clipPath').attr('id', 'clip_visidd600d1d0-9993-4f7c-b9b2-4f2380ec32f2_1').append('rect')\n",
       "      .attr('x', -1).attr('y', -1)\n",
       "      .attr('width', geom.inner_width+2).attr('height', geom.inner_height+2);\n",
       "    chart.append('text').attr('class', 'title header').text('Precipitation Trend for Chile').style('text-anchor', 'middle').attr('dy','0.8em')\n",
       "      .attr('x','50%')\n",
       "      .attr('y',0);\n",
       "\n",
       "    // Scales //////////////////////////////////////////////////////////////////////////////////////\n",
       "\n",
       "    var scale_x = d3.scale.ordinal()\n",
       "      .domain(['1990', '1995', '1996', '1997', '1998', '1999', '2000', '2001', '2002', '2003', '2004', \n",
       "        '2005', '2006', '2007'])\n",
       "      .rangePoints([0, geom.inner_width], 1);\n",
       "    var scale_inner = d3.scale.linear().domain([0,1])\n",
       "      .range([-0.5, 0.5]);\n",
       "    var scale_y = d3.scale.linear()\n",
       "      .domain([700000, 1400000])\n",
       "      .range([geom.inner_height,0]);\n",
       "\n",
       "    // Axes ////////////////////////////////////////////////////////////////////////////////////////\n",
       "\n",
       "    axes.append('g').attr('class', 'x axis')\n",
       "      .attr('transform','translate(0,' + geom.inner_height + ')');\n",
       "    axes.select('g.axis.x').append('text').attr('class', 'title')\n",
       "      .attr('text-anchor', 'middle')\n",
       "      .attr('x', geom.inner_width/2)\n",
       "      .attr('y', geom.inner_bottom - 6.0)\n",
       "      .text('Years');\n",
       "    axes.append('g').attr('class', 'y axis')\n",
       "      .attr('transform','translate(geom.chart_left, 0)');\n",
       "    axes.select('g.axis.y').append('text').attr('class', 'title')\n",
       "      .attr('text-anchor', 'middle')\n",
       "      .attr('x', -geom.inner_height/2)\n",
       "      .attr('y', 6-geom.inner_left).attr('dy', '0.7em').attr('transform', 'rotate(270)')\n",
       "      .text('Precipitation (million cubic meters)');\n",
       "\n",
       "    var axis_bottom = d3.svg.axis()\n",
       "      .scale(scale_x).innerTickSize(3).outerTickSize(0)\n",
       "      .ticks(Math.min(10, Math.round(geom.inner_width / 39.0)));\n",
       "    var axis_left = d3.svg.axis().orient('left')\n",
       "      .scale(scale_y).innerTickSize(3).outerTickSize(0);\n",
       "\n",
       "    function buildAxes() {\n",
       "      axes.select('g.axis.x').call(axis_bottom).selectAll('.tick text')\n",
       "        .style('text-anchor', 'end')\n",
       "        .attr('dx', '-.3em')\n",
       "        .attr('dy', '.6em')\n",
       "        .attr('transform', function(d) { return 'rotate(-45)' });\n",
       "      axes.select('g.axis.y').call(axis_left);\n",
       "    }\n",
       "\n",
       "    // Define element #1 ///////////////////////////////////////////////////////////////////////////\n",
       "\n",
       "    elements[0] = function() {\n",
       "      var original, processed,           // data sets passed in and then transformed\n",
       "        element, data,                   // Brunel element information and brunel data\n",
       "        selection;                       // D3 selection\n",
       "      var elementGroup = interior.append('g').attr('class', 'element1'),\n",
       "        main = elementGroup.append('g').attr('class', 'main'),\n",
       "        labels = BrunelD3.undoTransform(elementGroup.append('g').attr('class', 'labels'), elementGroup);\n",
       "\n",
       "      function makeData() {\n",
       "        original = datasets[0];\n",
       "        if (filterRows) original = original.retainRows(filterRows);\n",
       "        processed = pre(original, 0);\n",
       "        processed = post(processed, 0);\n",
       "        var f0 = processed.field('Year'),\n",
       "          f1 = processed.field('Precipitation'),\n",
       "          f2 = processed.field('#row');\n",
       "        var keyFunc = function(d) { return f0.value(d) };\n",
       "        data = {\n",
       "          Year:         function(d) { return f0.value(d.row) },\n",
       "          Precipitation:function(d) { return f1.value(d.row) },\n",
       "          $row:         function(d) { return f2.value(d.row) },\n",
       "          Year_f:       function(d) { return f0.valueFormatted(d.row) },\n",
       "          Precipitation_f:function(d) { return f1.valueFormatted(d.row) },\n",
       "          $row_f:       function(d) { return f2.valueFormatted(d.row) },\n",
       "          _split:       function(d) { return 'ALL' },\n",
       "          _key:         keyFunc,\n",
       "          _rows:        BrunelD3.makeRowsWithKeys(keyFunc, processed.rowCount())\n",
       "        };\n",
       "      }\n",
       "\n",
       "      // Build element from data ///////////////////////////////////////////////////////////////////\n",
       "\n",
       "      function build(transitionMillis) {\n",
       "        element = elements[0];\n",
       "        var x = function(d) { return scale_x(data.Year(d))};\n",
       "        var w = 0.9 * geom.inner_width/14;\n",
       "        var y = function(d) { return scale_y(data.Precipitation(d))};\n",
       "        var h = geom.default_point_size;\n",
       "        selection = main.selectAll('*').data(data._rows, function(d) { return d.key});\n",
       "        selection.enter().append('circle').attr('class',  'element point filled');\n",
       "\n",
       "        BrunelD3.trans(selection,transitionMillis)\n",
       "          .attr('cx',function(d) { return scale_x(data.Year(d))})\n",
       "          .attr('cy',function(d) { return scale_y(data.Precipitation(d))})\n",
       "          .attr('r',Math.min(0.9 * geom.inner_width/14, geom.default_point_size) / 2);\n",
       "\n",
       "        BrunelD3.trans(selection.exit(),transitionMillis/3)\n",
       "          .style('opacity', 0.5).each( function() {this.remove(); if (this.__label__) this.__label__.remove()});\n",
       "      }\n",
       "\n",
       "      return {\n",
       "        data:           function() { return processed },\n",
       "        internal:       function() { return data },\n",
       "        selection:      function() { return selection },\n",
       "        makeData:       makeData,\n",
       "        build:          build,\n",
       "        fields: {\n",
       "          x:            ['Year'],\n",
       "          y:            ['Precipitation'],\n",
       "          key:          ['Year']\n",
       "        }\n",
       "      };\n",
       "    }();\n",
       "\n",
       "    // Define element #2 ///////////////////////////////////////////////////////////////////////////\n",
       "\n",
       "    elements[1] = function() {\n",
       "      var original, processed,           // data sets passed in and then transformed\n",
       "        element, data,                   // Brunel element information and brunel data\n",
       "        selection;                       // D3 selection\n",
       "      var elementGroup = interior.append('g').attr('class', 'element2'),\n",
       "        main = elementGroup.append('g').attr('class', 'main'),\n",
       "        labels = BrunelD3.undoTransform(elementGroup.append('g').attr('class', 'labels'), elementGroup);\n",
       "\n",
       "      function makeData() {\n",
       "        original = datasets[0];\n",
       "        if (filterRows) original = original.retainRows(filterRows);\n",
       "        processed = pre(original, 0)\n",
       "          .sortRows('Year:ascending');\n",
       "        processed = post(processed, 0);\n",
       "        var f0 = processed.field('Year'),\n",
       "          f1 = processed.field('BestLine'),\n",
       "          f2 = processed.field('#row');\n",
       "        var keyFunc = function(d) { return 'ALL' };\n",
       "        data = {\n",
       "          Year:         function(d) { return f0.value(d.row) },\n",
       "          BestLine:     function(d) { return f1.value(d.row) },\n",
       "          $row:         function(d) { return f2.value(d.row) },\n",
       "          Year_f:       function(d) { return f0.valueFormatted(d.row) },\n",
       "          BestLine_f:   function(d) { return f1.valueFormatted(d.row) },\n",
       "          $row_f:       function(d) { return f2.valueFormatted(d.row) },\n",
       "          _split:       function(d) { return 'ALL' },\n",
       "          _key:         keyFunc,\n",
       "          _rows:        BrunelD3.makeRowsWithKeys(keyFunc, processed.rowCount())\n",
       "        };\n",
       "      }\n",
       "\n",
       "      // Build element from data ///////////////////////////////////////////////////////////////////\n",
       "\n",
       "      function build(transitionMillis) {\n",
       "        element = elements[1];\n",
       "        var x = function(d) { return scale_x(data.Year(d))};\n",
       "        var w = 0.9 * geom.inner_width/14;\n",
       "        var y = function(d) { return scale_y(data.BestLine(d))};\n",
       "        var h = function(d) { return Math.abs( scale_y(5451.913052050397) - scale_y(0) )};\n",
       "        var path = d3.svg.line().x(x).y(y);\n",
       "        var splits = BrunelD3.makePathSplits(data, path, x);\n",
       "        selection = main.selectAll('*').data(splits, function(d) { return d.key});\n",
       "        selection.enter().append('path').attr('class',  'element line');\n",
       "\n",
       "        BrunelD3.trans(selection,transitionMillis)\n",
       "          .attr('d', function(d) { return d.path });\n",
       "\n",
       "        BrunelD3.trans(selection.exit(),transitionMillis/3)\n",
       "          .style('opacity', 0.5).each( function() {this.remove(); if (this.__label__) this.__label__.remove()});\n",
       "      }\n",
       "\n",
       "      return {\n",
       "        data:           function() { return processed },\n",
       "        internal:       function() { return data },\n",
       "        selection:      function() { return selection },\n",
       "        makeData:       makeData,\n",
       "        build:          build,\n",
       "        fields: {\n",
       "          x:            ['Year'],\n",
       "          y:            ['BestLine']\n",
       "        }\n",
       "      };\n",
       "    }();\n",
       "\n",
       "    function build(time) {\n",
       "      var first = elements[0].data() == null;\n",
       "      if (first) time = 0; // No transition for first call\n",
       "      buildAxes();\n",
       "      if (first || time > -1) {\n",
       "        elements[0].makeData();\n",
       "        elements[1].makeData();\n",
       "      }\n",
       "      elements[0].build(time);\n",
       "      elements[1].build(time);\n",
       "    }\n",
       "\n",
       "    // Expose the following components of the chart\n",
       "    return {build : build, elements : elements, scales: {x:scale_x, y:scale_y}};\n",
       "    }();\n",
       "\n",
       "  function setData(rowData, i) { datasets[i||0] = BrunelD3.makeData(rowData) }\n",
       "  function updateAll(time) { charts.forEach(function(x) {x.build(time || 0)}) }\n",
       "  function buildAll() {\n",
       "    for (var i=0;i<arguments.length;i++) setData(arguments[i], i);\n",
       "    updateAll(transitionTime);\n",
       "  }\n",
       "\n",
       "  return {\n",
       "    dataPreProcess:     function(f) { if (f) pre = f; return pre },\n",
       "    dataPostProcess:    function(f) { if (f) post = f; return post },\n",
       "    data:               function(d,i) { if (d) setData(d,i); return datasets[i||0] },\n",
       "    visId:              visId,\n",
       "    build:              buildAll,\n",
       "    rebuild:            updateAll,\n",
       "    charts:             charts\n",
       "  }\n",
       "}\n",
       "\n",
       "// Data Tables /////////////////////////////////////////////////////////////////////////////////////\n",
       "\n",
       "var table1 = {\n",
       "   names: ['Year', 'Precipitation', 'BestLine'], \n",
       "   options: ['string', 'numeric', 'numeric'], \n",
       "   rows: [['1998', 951060, 1120906.78704512], ['1990', 1152225, 1077291.48262872],\n",
       "  ['2000', 1160289, 1131810.61314922], ['2002', 1306221, 1142714.43925332],\n",
       "  ['2006', 1181614, 1164522.09146152], ['2004', 1272096.625, 1153618.26535742],\n",
       "  ['1996', 897224, 1110002.96094102], ['2007', 800225.1875, 1169974.00451357],\n",
       "  ['2005', 1231319.25, 1159070.17840947], ['1995', 971392, 1104551.04788897],\n",
       "  ['1997', 1315783, 1115454.87399307], ['2001', 1239664, 1137262.52620127],\n",
       "  ['2003', 1377889, 1148166.35230537], ['1999', 1083755, 1126358.70009717]]\n",
       "};\n",
       "\n",
       "// Call Code to Build the system ///////////////////////////////////////////////////////////////////\n",
       "\n",
       "var v = new BrunelVis('visidd600d1d0-9993-4f7c-b9b2-4f2380ec32f2');\n",
       "v.build(table1);\n",
       "\n",
       "            \"\"\n",
       "        });\n",
       "        });\n",
       "        </script>"
      ]
     },
     "execution_count": 64,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%brunel\n",
    "data(\"precipBestLineDF\") title(\"Precipitation Trend for Chile\") \n",
    "x(Year:linear) y(Precipitation) \n",
    "axes(x:'Years',y:'Precipitation (million cubic meters)') interaction(none)\n",
    "+ line x(Year:linear) y(BestLine) axes(x:'Years',y:'Precipitation (million cubic meters)') interaction(none)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "NOTE: Can't adjust the unit of spacing between ticks on x axis. Therefore, the line of best fit has a little curve."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For Chile, we observe a positive trend in annual precipitation, despite the fact that during the last couple of years the annual precipitation decreased. This example shows you how to determine the trend for the annual precipitation of one country. Now we have to do it for all 91 countries or areas. Let us implement an automated way to determine the trends for all countries and areas."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "totalPrecipByCountryDF.show(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We define a function that accepts a row holding precipatition values and the years measures were taken. It filters the years when the precipitation was zero. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def filterZeroValuesAndAssociatedYears(row:Array[Double],years:Array[String]):Tuple2[List[String],List[Double]] = {\n",
    "   \n",
    "   var yearRow = (years,row).zipped\n",
    "   var t:List[Tuple2[String,Double]] = yearRow.map(\n",
    "   (x,y)=>new Tuple2(x,y)).toList\n",
    "   \n",
    "    var filtered = t.filter(r=>r._2>0)\n",
    "   \n",
    "    var year = filtered.map(r=>r._1)\n",
    "    var precip = filtered.map(r=>r._2)\n",
    "   \n",
    "   return (year,precip)\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "var iterator = df.drop(\"Country or Area\").rdd.toLocalIterator\n",
    "var countries = df.select(\"Country or Area\").rdd.collect().map(r=>r(0))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we iterate through each row and pass it to the filterZeroValuesAndAssoicatedYears function which preprocesses the data before it is past to the computeSlope function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import scala.collection.mutable.ArrayBuffer\n",
    "var slopes = ArrayBuffer[Double]()\n",
    "\n",
    "while(iterator.hasNext)\n",
    "{\n",
    "    val row = iterator.next()\n",
    "    var filteredPair = filterZeroValuesAndAssociatedYears(row.toSeq.toArray.map(_.asInstanceOf[Double]),years)\n",
    "    slopes += computeSlope(filteredPair._1.toArray.map(_.toDouble),filteredPair._2.toArray)\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var slopeByCountry = (countries,slopes).zipped"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we have an Array of Tuples, containing the slope of the fitted lines. If the value is positive, there is a positive trend for annual precipitation. If the value is negative, there is a negative trend for annual precipitation. Values near zero indicate a stable condition. \n",
    "\n",
    "To answer our second question, we have to determine all v with a negative value for the slope. The corresponding code is in the cell below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Andorra\n",
      "Anguilla\n",
      "Bahrain\n",
      "Barbados\n",
      "Central African Republic\n",
      "China\n",
      "China, Hong Kong SAR\n",
      "Cuba\n",
      "Denmark\n",
      "Estonia\n",
      "Guinea\n",
      "Iraq\n",
      "Israel\n",
      "Italy\n",
      "Kazakhstan\n",
      "Latvia\n",
      "Lithuania\n",
      "Luxembourg\n",
      "Marshall Islands\n",
      "Monaco\n",
      "Morocco\n",
      "Paraguay\n",
      "Poland\n",
      "Portugal\n",
      "Qatar\n",
      "Republic of Moldova\n",
      "Slovenia\n",
      "Spain\n",
      "Switzerland\n",
      "Togo\n",
      "Tunisia\n",
      "Venezuela\n",
      "Yemen\n"
     ]
    }
   ],
   "source": [
    "var negativeTrendSlope = slopeByCountry.map((x,y)=> Tuple2(x,y)).filter(r=>r._2<0)\n",
    "var negativeTrendCountries  = negativeTrendSlope.map(r=>println(r._1))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Question 3: Which country or area diplays the steepest positive trend in precipitation?\n",
    "We leave this question for you to be answered."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Download Notebooks\n",
    "After performing an analysis, you can download your results. In the **Menu Bar**, go to **File** and then to **Download as**. It is possible to download notebooks in various formats to your local file system. Then you can send your work and results to colleagues."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "(C) Copyright IBM Corp. 2016"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Scala 2.10",
   "language": "scala",
   "name": "scala"
  },
  "language_info": {
   "name": "scala"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}