HW5.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's import and initialize Spark, along with all the other libraries we're using (this might take a while)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Uncomment these if do this homework locally\n",
    "# See Homework 3 for details\n",
    "\n",
    "# from local_install import setup_environment\n",
    "# setup_environment() \n",
    "\n",
    "import folium\n",
    "import pyspark\n",
    "from pyspark.sql.types import *\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "import os\n",
    "%pylab inline\n",
    "%pylab notebook"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = pyspark.SparkContext()\n",
    "sql = pyspark.sql.SQLContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 1. DataFrames"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1a. File Format Wrangling"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Begin Student Code Here"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Implement the loading of data files into DataFrame objects. First, unzip the data file into a local folder whose path is stored in 'root_path'."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "root_path = \"./fec_2016_3_25\"\n",
    "\n",
    "# file_struct objects\n",
    "com_file = {'header': \"cm_header_file.csv\", 'data': \"cm.txt\" }\n",
    "can_file = {'header': \"cn_header_file.csv\", 'data': \"cn.txt\" }\n",
    "com_can_link_file = {'header': \"ccl_header_file.csv\", 'data': \"ccl.txt\" }\n",
    "indv_file = {'header': \"indiv_header_file.csv\", 'data': \"itcont.txt\" }\n",
    "pas_file = {'header': \"pas2_header_file.csv\", 'data': \"itpas2.txt\" }\n",
    "com_links_file = {'header': \"oth_header_file.csv\", 'data': \"itoth.txt\" }\n",
    "\n",
    "def load_header(filename):\n",
    "    \"\"\"\n",
    "    Given a header .csv file, return a list containing the names of all columns in the table.\n",
    "    \n",
    "    Input:\n",
    "    filename: a string specifying the header .csv file to load\n",
    "    \n",
    "    Output:\n",
    "    A list containing column names of the table\n",
    "    \"\"\"\n",
    "    with open(filename, \"r\") as f:\n",
    "        return [r.replace('\\r\\n','') for r in f.readline().split(\",\")]\n",
    "    \n",
    "def load_dataframe(file_struct):\n",
    "    \"\"\"\n",
    "    Given a dictionary representing the locations of FEC raw files corresponding to a table,\n",
    "    load the tables into a Spark DataFrame.\n",
    "    \n",
    "    Input:\n",
    "    file_struct: a dictionary containing the keys 'header' and 'data', where\n",
    "     'header' contains the name of the `.csv` file specifying the table header file, and\n",
    "     'data' contains the name of the `.txt` file specifying the table data file\n",
    "     \n",
    "    Output:\n",
    "    A DataFrame which contains the loaded data.\n",
    "    \"\"\"\n",
    "    df = None\n",
    "    # TODO\n",
    "    \n",
    "    return df.cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### End Student Code Here"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll load these files into Spark now, and register them as temporary SQL tables."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "com = load_dataframe(com_file)\n",
    "can = load_dataframe(can_file)\n",
    "links = load_dataframe(com_can_link_file)\n",
    "indv = load_dataframe(indv_file)\n",
    "pas = load_dataframe(pas_file)\n",
    "com_links = load_dataframe(com_links_file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "com.registerTempTable(\"com\")\n",
    "can.registerTempTable(\"can\")\n",
    "links.registerTempTable(\"links\")\n",
    "indv.registerTempTable(\"indv\")\n",
    "pas.registerTempTable(\"pas\")\n",
    "com_links.registerTempTable(\"com_links\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1b. Basic Analytics Queries"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As a demonstration, let's query for information corresponding to the strings \"Clinton\", \"Sanders\", \"Trump\", and \"Cruz\". Try these queries. How does this output look?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "front_runners = can.where(can.CAND_NAME.like(\"%CLINTON%\") | can.CAND_NAME.like(\"%SANDERS%\") \n",
    "          | can.CAND_NAME.like(\"%TRUMP%\") | can.CAND_NAME.like(\"%CRUZ%\") )\\\n",
    "    .select(\"CAND_ID\", \"CAND_NAME\", \"CAND_STATUS\", \"CAND_OFFICE\")\n",
    "front_runners.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "front_runners = sql.sql(\"\"\"\n",
    "SELECT CAND_ID, CAND_NAME, CAND_STATUS, CAND_OFFICE\n",
    "FROM can\n",
    "WHERE CAND_NAME LIKE \"%CLINTON%\"\n",
    "OR CAND_NAME LIKE \"%SANDERS%\"\n",
    "OR CAND_NAME LIKE \"%TRUMP%\"\n",
    "OR CAND_NAME LIKE \"%CRUZ%\"\n",
    "\"\"\")\n",
    "front_runners.registerTempTable(\"fr\")\n",
    "front_runners.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "That didn't work out so well. But notice how both queries got the same results! For the rest of the questions, you are allowed to write the queries in either form. Use the DataFrames that you created in part 1a. They are listed again here for reference:\n",
    "\n",
    "* **com** contains committee information [Details](http://www.fec.gov/finance/disclosure/metadata/DataDictionaryCommitteeMaster.shtml)\n",
    "* **can** contains candidate information [Details](http://www.fec.gov/finance/disclosure/metadata/DataDictionaryCandidateMaster.shtml)\n",
    "* **links** contains linkage between committees and candidates [Details](http://www.fec.gov/finance/disclosure/metadata/DataDictionaryCandCmteLinkage.shtml)\n",
    "* **indv** contains individual contributions to committees [Details](http://www.fec.gov/finance/disclosure/metadata/DataDictionaryContributionsbyIndividuals.shtml)\n",
    "* **pas** contains contributions between committees [Details](http://www.fec.gov/finance/disclosure/metadata/DataDictionaryContributionstoCandidates.shtml)\n",
    "* **com_links** links between committees [Details](http://www.fec.gov/finance/disclosure/metadata/DataDictionaryCommitteetoCommittee.shtml)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Begin Student Code Here"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Q1."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "front_runners = None\n",
    "# TODO\n",
    "\n",
    "q1 = front_runners.select(\"CAND_ID\", \"CAND_NAME\", \"CAND_PCC\")\n",
    "q1.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q1 Expected Output:**\n",
    "```\n",
    "+---------+-------------------------+---------+\n",
    "|CAND_ID  |CAND_NAME                |CAND_PCC |\n",
    "+---------+-------------------------+---------+\n",
    "|P00003392|CLINTON, HILLARY RODHAM  |C00575795|\n",
    "|P60006111|CRUZ, RAFAEL EDWARD \"TED\"|C00574624|\n",
    "|P60007168|SANDERS, BERNARD         |C00577130|\n",
    "|P80001571|TRUMP, DONALD J          |C00580100|\n",
    "+---------+-------------------------+---------+\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Q2."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "num_indv_contributions = None\n",
    "# TODO\n",
    "\n",
    "q2 = num_indv_contributions.select(\"CAND_ID\", \"CAND_NAME\", \"count\")\n",
    "q2.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q2 Expected Output:**\n",
    "```\n",
    "+---------+-------------------------+-----+\n",
    "|CAND_ID  |CAND_NAME                |count|\n",
    "+---------+-------------------------+-----+\n",
    "|P80001571|TRUMP, DONALD J          |3006 |\n",
    "|P60006111|CRUZ, RAFAEL EDWARD \"TED\"|34699|\n",
    "|P60007168|SANDERS, BERNARD         |34524|\n",
    "|P00003392|CLINTON, HILLARY RODHAM  |71249|\n",
    "+---------+-------------------------+-----+\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Q3."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "indv_contributions_amt = None\n",
    "# TODO\n",
    "\n",
    "# We will also accept a solution in either format.\n",
    "# Be sure to comment out the one you don't use: \n",
    "# q3 = indv_contributions_amt.select(\"CAND_ID\", \"CAND_NAME\", \"SUM\")\n",
    "q3 = indv_contributions_amt.select(\"CAND_ID\", \"CAND_NAME\", \"sum(TRANSACTION_AMT)\")\n",
    "q3.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q3 Expected Output:**\n",
    "```\n",
    "+---------+-------------------------+--------------------+\n",
    "|CAND_ID  |CAND_NAME                |sum(TRANSACTION_AMT)|\n",
    "+---------+-------------------------+--------------------+\n",
    "|P80001571|TRUMP, DONALD J          |1994976.0           |\n",
    "|P60006111|CRUZ, RAFAEL EDWARD \"TED\"|2.4785175E7         |\n",
    "|P60007168|SANDERS, BERNARD         |4.9001141E7         |\n",
    "|P00003392|CLINTON, HILLARY RODHAM  |9.5996062E7         |\n",
    "+---------+-------------------------+--------------------+\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Q4."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "linked_committees = None\n",
    "# TODO\n",
    "\n",
    "q4 = linked_committees.select(\"CAND_NAME\", \"CAND_ID\", \"CMTE_ID\", \"CMTE_NM\")\n",
    "q4.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q4 Expected Output:**\n",
    "```\n",
    "+-------------------------+---------+---------+-------------------------------------------------------------------------------+\n",
    "|CAND_NAME                |CAND_ID  |CMTE_ID  |CMTE_NM                                                                        |\n",
    "+-------------------------+---------+---------+-------------------------------------------------------------------------------+\n",
    "|CLINTON, HILLARY RODHAM  |P00003392|C00577395|PEOPLE IN COMMAND/PIC                                                          |\n",
    "|SANDERS, BERNARD         |P60007168|C00590646|NEW YORK CAPITAL REGION FOR BRINGING ECONOMIC REVOLUTION NOW INSPIRING EVERYONE|\n",
    "|SANDERS, BERNARD         |P60007168|C00612549|LORAIN COUNTY FORWARD                                                          |\n",
    "|CLINTON, HILLARY RODHAM  |P00003392|C00575795|HILLARY FOR AMERICA                                                            |\n",
    "|SANDERS, BERNARD         |P60007168|C00587766|SILVER CITY NM FOR BERNIE SANDERS                                              |\n",
    "|SANDERS, BERNARD         |P60007168|C00588707|ITHACA AND TOMPKINS COUNTY FOR BERNIE SANDERS                                  |\n",
    "|SANDERS, BERNARD         |P60007168|C00589937|PROGRESSGJ                                                                     |\n",
    "|SANDERS, BERNARD         |P60007168|C00588434|ROCHESTER FOR PROGRESS                                                         |\n",
    "|SANDERS, BERNARD         |P60007168|C00577130|BERNIE 2016                                                                    |\n",
    "|CRUZ, RAFAEL EDWARD \"TED\"|P60006111|C00574624|CRUZ FOR PRESIDENT                                                             |\n",
    "|SANDERS, BERNARD         |P60007168|C00590828|UPPER VALLEY FOR BERNIE SANDERS                                                |\n",
    "|CRUZ, RAFAEL EDWARD \"TED\"|P60006111|C00612119|CRUZ INFO PRESIDENT                                                            |\n",
    "|SANDERS, BERNARD         |P60007168|C00583708|PROGRESS WV                                                                    |\n",
    "|SANDERS, BERNARD         |P60007168|C00582395|UPSTATE NEW YORK FOR BERNIE SANDERS                                            |\n",
    "|SANDERS, BERNARD         |P60007168|C00590620|LAS CRUCES FOR BERNIE                                                          |\n",
    "|CLINTON, HILLARY RODHAM  |P00003392|C00570978|FUTURE OF AMERICAN LIVES MATTER                                                |\n",
    "|TRUMP, DONALD J          |P80001571|C00580100|DONALD J. TRUMP FOR PRESIDENT, INC.                                            |\n",
    "|SANDERS, BERNARD         |P60007168|C00590240|NORTHEAST MUSICIANS FOR SOCIAL DEMOCRACY                                       |\n",
    "+-------------------------+---------+---------+-------------------------------------------------------------------------------+\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Q5."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "num_com_contributions = None\n",
    "# TODO\n",
    "\n",
    "q5 = num_com_contributions.select(\"CAND_NAME\", \"count\")\n",
    "q5.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q5 Expected Output:**\n",
    "\n",
    "```\n",
    "+-------------------------+-----+\n",
    "|CAND_NAME                |count|\n",
    "+-------------------------+-----+\n",
    "|CLINTON, HILLARY RODHAM  |2561 |\n",
    "|SANDERS, BERNARD         |1577 |\n",
    "|TRUMP, DONALD J          |333  |\n",
    "|CRUZ, RAFAEL EDWARD \"TED\"|643  |\n",
    "+-------------------------+-----+\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "Q6."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "com_contributions_amt = None\n",
    "# TODO\n",
    "\n",
    "# We will also accept a solution in either format.\n",
    "# Be sure to comment out the one you don't use: \n",
    "# q6 = com_contributions_amt.select(\"CAND_NAME\", \"SUM\")\n",
    "q6 = com_contributions_amt.select(\"CAND_NAME\", \"sum(TRANSACTION_AMT)\")\n",
    "q6.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q6 Expected Output:**\n",
    "\n",
    "```\n",
    "+-------------------------+--------------------+\n",
    "|CAND_NAME                |sum(TRANSACTION_AMT)|\n",
    "+-------------------------+--------------------+\n",
    "|CLINTON, HILLARY RODHAM  |6064499.0           |\n",
    "|SANDERS, BERNARD         |1202034.0           |\n",
    "|TRUMP, DONALD J          |6099777.0           |\n",
    "|CRUZ, RAFAEL EDWARD \"TED\"|1.2145601E7         |\n",
    "+-------------------------+--------------------+\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### End Student Code Here"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2. K-means Clustering"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## Use Dataframe API to load some Toy Data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Load a parquet file as a DataFrame."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = sql.read.parquet(\"toy_data\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's look at a few records:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Plotting the Toy Data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll first plot a sample of some toy data. Let's collect a sample of records,"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample = df.sample(False, 0.2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "project out the `x` and the `y` columns,"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "x_sample = sample.select('x').collect()\n",
    "y_sample = sample.select('y').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "and plot the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "plt.figure()\n",
    "plt.axis([-10, 10, -10, 10])\n",
    "plot(x_sample, y_sample, 'bo')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Implementation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Implement a distributed version of K-Means clustering using PySpark."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2a. K-Means++ Initialization"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Initialize the centers using the K-Means++ Algorithm, using Distributed Reservoir Sampling. We've provided you the signatures of a few functions which may be helpful."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Begin Student Code Here"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def initialize_centers_plus(points, k):\n",
    "    \"\"\"\n",
    "    Find k initial cluster centers using distributed weighted reservoir sampling.\n",
    "    \n",
    "    Inputs:\n",
    "    points: a collection of d-dimensional points (x_1, x_2, ..., x_d).\n",
    "    k: the number of cluster centers wanted\n",
    "    \n",
    "    Output:\n",
    "    A list of k points which become the initial centers in K-means clustering\n",
    "    \"\"\"\n",
    "    centers = []\n",
    "    \n",
    "    # Choose first center uniformly at random\n",
    "    # TODO\n",
    "    \n",
    "    for _ in range(1, k):\n",
    "        # Compute distances of each point to its nearest center, and\n",
    "        # TODO\n",
    "        \n",
    "        # given distances, choose a new center\n",
    "        # TODO\n",
    "        \n",
    "        \n",
    "    return centers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def choose_partition_center(points_partition_iterator):\n",
    "    \"\"\"\n",
    "    Choose a single center from a SINGLE PARTITION, using weighted random sampling.\n",
    "    \n",
    "    Inputs:\n",
    "    points_partition_iterator: an iterator through a partition of D-dimensional points\n",
    "    \n",
    "    Output:\n",
    "    (As an iterator) A 2-tuple, containing the randomly-chosen center and its weighted random-sampling key\n",
    "    \"\"\"\n",
    "    center = None\n",
    "    key = None\n",
    "    # TODO\n",
    "    \n",
    "    \n",
    "    yield (center, key)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def pick_between_centers(candidate_center_1, candidate_center_2):\n",
    "    \"\"\"\n",
    "    Between two centers from different partitions, pick the one with a larger key.\n",
    "    \n",
    "    Inputs:\n",
    "    candidate_center_1: a 2-tuple, containing the randomly-chosen center and its weighted random sampling key \n",
    "                        from the first partition\n",
    "    candidate_center_2: a 2-tuple, containing the randomly-chosen center and its weighted random sampling key \n",
    "                        from the first partition\n",
    "    \n",
    "    Output:\n",
    "    A 2-tuple, containing the \"better\" center and its weighted random-sampling key\n",
    "    \"\"\"\n",
    "    better_center = None\n",
    "    # TODO\n",
    "    \n",
    "    \n",
    "    return better_center"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def nearest_center(centers, point):\n",
    "    \"\"\"\n",
    "    Given a point and a list of centers (also points), determine the center closest to that point,\n",
    "    and compute the distance to that point.\n",
    "    \n",
    "    Inputs:\n",
    "    centers: a list of points which represent the current centers\n",
    "    point: the point to examine\n",
    "    \n",
    "    Outputs:\n",
    "    A 2-tuple, containing the index of the closest center (point) and its distance from point\n",
    "    \"\"\"\n",
    "    (index, shortest_distance) = (None, None)\n",
    "    # TODO\n",
    "    \n",
    "    \n",
    "    return (index, shortest_distance)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### End Student Code Here"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Plotting the centers"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's see how our initialization algorithm performs. How do your centers look? Are they too close to each other?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "num_centers = 4\n",
    "centers = initialize_centers_plus(df, num_centers)\n",
    "print(centers) \n",
    "\n",
    "plt.figure()\n",
    "plt.axis([-10, 10, -10, 10])\n",
    "plot(x_sample, y_sample, 'bo')\n",
    "centers = np.asarray(centers)\n",
    "plt.plot(centers[:,0], centers[:,1], 'mo', markersize = 10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2b. Main Loop"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, implement the main loop in K-means clustering. Again, we've provided you the signatures of some functions which may be useful."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note: The statistics of a new center \"mean\" is given by:\n",
    "$$\n",
    "\\bar{x}_k = \\frac{1}{n_k} \\sum_{x \\in \\text{Cluster}[k]}^{n_i} x\n",
    "$$\n",
    "So the only statistics we require are $n_k$ the number of elements in cluster $k$ and $\\sum_{x \\in \\text{Cluster}[k]}^{n_i} x$ the sum of the elements in cluster $k$."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Begin Student Code Here"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def k_means(points,\n",
    "            k = num_centers,\n",
    "            initial_centers = None,\n",
    "            max_iterations = 100,\n",
    "            initializer = initialize_centers_plus,\n",
    "            epsilon = 0.001):\n",
    "    \"\"\"\n",
    "    Executes the K-means algorithm on a collection of points.\n",
    "    \n",
    "    Inputs:\n",
    "    points: a collection of d-dimensional points (x_1, x_2, ..., x_d).\n",
    "    k: the number of cluster centers wanted\n",
    "    initial_centers: if supplied, skips the initialization phase and uses points from this value\n",
    "    max_iterations: the maximum number of main-loop iterations to run\n",
    "    initializer: a function which selects initial centers (if none supplied)\n",
    "    epsilon: the threshhold at which convergence is reached and the algorithm halted\n",
    "    \n",
    "    Output:\n",
    "    A list of k candidate cluster centers\n",
    "    \"\"\"\n",
    "    \n",
    "    # speeds up rerunning\n",
    "    points.cache()\n",
    "    \n",
    "    old_centers = None\n",
    "    new_centers = None\n",
    "    if initial_centers: # we were provided initial centers to use\n",
    "        new_centers = initial_centers\n",
    "    else: # we need to initialize the centers ourselves\n",
    "        # TODO\n",
    "\n",
    "        \n",
    "    iteration = 0\n",
    "    while not has_converged(old_centers, new_centers, epsilon) and iteration < max_iterations:\n",
    "        # update centers\n",
    "        # TODO\n",
    "        \n",
    "        \n",
    "        iteration += 1\n",
    "        \n",
    "    return new_centers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def compute_new_center_statistics(iterator, old_centers):\n",
    "    \"\"\"\n",
    "    Given an iterator over points and a list of old centers, compute the statistics for the new center.\n",
    "    \n",
    "    Input:\n",
    "    iterator: an iterator over points\n",
    "    old_centers: a list of centers (points) from the previous iteration\n",
    "    \n",
    "    Output:\n",
    "    A 2-tuple (counts, sums) consisting of:\n",
    "     - counts: an array of length k containing the count of points in each new center\n",
    "     - sums: a k by d array consisting of sum of the points assigned to each center\n",
    "     \n",
    "    Note that from the tuple you could compute the ith new center:\n",
    "      sums[i] / counts[i]\n",
    "    \"\"\"\n",
    "    # Get the shape of the old centers\n",
    "    # TODO\n",
    "    \n",
    "    # Initialize the sums\n",
    "    # TODO\n",
    "    counts = None\n",
    "    sums = None\n",
    "    \n",
    "    # Loop over the data and compute the new assignments\n",
    "    # TODO\n",
    "    for _ in []:\n",
    "        # Compute the nearest center (you just implemented this!)\n",
    "        # TODO\n",
    "        \n",
    "        # Update the sums and counts\n",
    "        # TODO\n",
    "        \n",
    "        pass\n",
    "\n",
    "\n",
    "        \n",
    "    yield (counts, sums)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def add_statistics(stats_1, stats_2):\n",
    "    \"\"\"\n",
    "    Given statistics from two partitions, add those statistics.\n",
    "    \"\"\"\n",
    "    stats = (None, None)\n",
    "    # Compute stats_1 + stats_2\n",
    "    # TODO\n",
    "    \n",
    "    return stats"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### End Student Code Here"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def has_converged(old_centers, new_centers, epsilon):\n",
    "    \"\"\"\n",
    "    Test if the distance between the centers is less than epsilon.\n",
    "    \"\"\"\n",
    "    return (old_centers is not None) and (new_centers is not None) and \\\n",
    "        np.linalg.norm(np.asarray(old_centers) - np.asarray(new_centers), ord=2) < epsilon"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Plotting Clusters"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "How does your code perform? Are these good clusters?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "colors = ['ro', 'bo', 'go', 'yo'] # for now just use 4 colors\n",
    "\n",
    "def plot_clusters(centers, plt):\n",
    "    for i in range(num_centers):\n",
    "        cluster = df.rdd.filter(lambda x: nearest_center(centers, x)[0] == i)\n",
    "        cluster_sample = np.asarray(cluster.sample(False, 0.2).collect())\n",
    "        plt.plot(cluster_sample[:,0], cluster_sample[:,1], colors[i])\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plt.figure()\n",
    "plt.axis([-10, 10, -10, 10])\n",
    "\n",
    "centers = k_means(df, num_centers)\n",
    "plot_clusters(centers, plt)\n",
    "\n",
    "centers = np.asarray(centers)\n",
    "plt.plot(centers[:,0], centers[:,1], 'mo', markersize = 10)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 3. Geographical Contribution Clustering"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's put everything together! First we'll load the required data files."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "zip_codes = sql.read.parquet(\"zip_codes\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we'll grab the relevant data we need. We'll join the table of front runners we collected earlier with their contributors' zip codes to get latitudinal and longitudinal data..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "contrib_zips = \\\n",
    "    front_runners.join(indv, front_runners.CAND_PCC == indv.CMTE_ID) \\\n",
    "    .select(\"CAND_NAME\", \"CAND_ID\", \"ZIP_CODE\")\n",
    "contrib_zips = contrib_zips.withColumn(\"SHORT_ZIP\", contrib_zips.ZIP_CODE.substr(0,5))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "contrib_locs = \\\n",
    "    contrib_zips.join(zip_codes, contrib_zips.SHORT_ZIP == zip_codes.zip) \\\n",
    "    .select(\"CAND_NAME\", \"latitude\", \"longitude\").cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "candidate_names = [x[0] for x in contrib_locs.select(\"CAND_NAME\").distinct().collect()]\n",
    "candidate_names"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "X = {}\n",
    "for candidate in candidate_names:\n",
    "    X[candidate] = contrib_locs.where(contrib_locs.CAND_NAME == candidate) \\\n",
    "        .select(\"latitude\", \"longitude\") \\\n",
    "        .rdd.map(lambda r: np.array([x for x in r])) \\\n",
    "        .coalesce(sc.defaultParallelism).cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "...which will allow us to plot these locations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "maps = {}\n",
    "for candidate in X:\n",
    "    maps[candidate] = folium.Map(location=[40, -100],\n",
    "           tiles='Stamen Toner',\n",
    "           zoom_start=4)\n",
    "    locs = X[candidate].sample(False, 0.01).collect()\n",
    "    for l in locs:\n",
    "        folium.Marker(l).add_to(maps[candidate])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "maps[u'CRUZ, RAFAEL EDWARD \"TED\"']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can run k-means for each candidate. What's a good value for $k$? Run the K-means algorithm you developed earlier with different values of $k$ to determine the best one."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Begin Student Code Here"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def compute_errors(candidate_locations, k_range):\n",
    "    errors = {}\n",
    "    for c in candidate_locations:\n",
    "        errors[c] = []\n",
    "        for k in k_range:\n",
    "            # For each candidate, test multiple values of k\n",
    "            # TODO\n",
    "            \n",
    "            pass\n",
    "    return errors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plt.figure()\n",
    "plt.axis([0, 30, 0, 20])\n",
    "k_range = range(2, 32, 4)\n",
    "errors = compute_errors(X, k_range)\n",
    "for c in errors:\n",
    "    plot(k_range, errors[c])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "Now that you've found a suitable value for k, run K-means to compute the desired clusters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "centers = {}\n",
    "k = None # TODO enter the value you found here\n",
    "for c in X:\n",
    "    # Find the cluster centers for each candidate\n",
    "    # TODO\n",
    "    pass"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### End Student Code Here"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At last, we have the clusters we want. Here's a nifty visualization."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "colors = dict(zip(centers.keys(), ['blue', 'green','orange', 'red']))\n",
    "centers_map = folium.Map(location=[40, -100],\n",
    "                         zoom_start=4)\n",
    "for candidate in centers: \n",
    "    locs = centers[candidate]\n",
    "    for l in locs:\n",
    "        folium.Marker(l,\n",
    "                      popup=str(l),\n",
    "                      icon=folium.Icon(color=colors[candidate])).add_to(centers_map)\n",
    "centers_map"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Congratulations! You've successfully implemented a scalable machine learning algorithm on dataset of campaign finance contributions!"
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Raw Cell Format",
  "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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}