school_donations_EMR.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "Current session configs: <tt>{'conf': {'spark.pyspark.python': 'python3', 'spark.pyspark.virtualenv.enabled': 'true', 'spark.pyspark.virtualenv.type': 'native', 'spark.pyspark.virtualenv.bin.path': '/usr/bin/virtualenv'}, 'kind': 'pyspark'}</tt><br>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "No active sessions."
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%info"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "c33092f53bf54b259e6ae281bf71e589",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>0</td><td>application_1576014270975_0001</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-172-31-26-120.us-west-2.compute.internal:20888/proxy/application_1576014270975_0001/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-172-31-20-195.us-west-2.compute.internal:8042/node/containerlogs/container_1576014270975_0001_01_000001/livy\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<SparkContext master=yarn appName=livy-session-0>"
     ]
    }
   ],
   "source": [
    "sc"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Set up"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "3b13c67d0b854ffa99c97a33464b7399",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "donations_rdd = sc.textFile('s3://dcgroup11/data/Donations.csv')\n",
    "donors_rdd = sc.textFile('s3://dcgroup11/data/Donors.csv')\n",
    "small_projects_rdd = sc.textFile('s3://dcgroup11/data/ProjectsSmall.csv')\n",
    "schools_rdd = sc.textFile('s3://dcgroup11/data/Schools.csv')\n",
    "teachers_rdd = sc.textFile('s3://dcgroup11/data/Teachers.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "7ec75844344349189abc2a6d9d784a51",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# The access keys have been removed to protect the account."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "68256f1e4dfe4947bf1f33209d2e6f67",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def remove_exist(name):\n",
    "    \"\"\"This removes an existing directory.\"\"\"\n",
    "    if os.path.exists(name):\n",
    "        shutil.rmtree(name)\n",
    "        \n",
    "# output_folder_name = 's3a://dcgroup11/data/donor_donation.csv'\n",
    "        \n",
    "# remove_exist(output_folder_name)  # Old folder is removed, if exists.\n",
    "\n",
    "# merged_donations.saveAsTextFile(output_folder_name)  # Here will save file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "a35bc98a367048ce912f38a890de56c6",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def rid_titles(rdd):\n",
    "    \"\"\"\n",
    "    Get rid of the title row for each rdd.\n",
    "    \"\"\"\n",
    "    header = rdd.first()\n",
    "    content = rdd.filter(lambda line: line != header).map(lambda x: x.split(','))\n",
    "    header_list = header.split(',')\n",
    "    return header_list, content\n",
    "\n",
    "def read_file(file_name):\n",
    "    \"\"\"\n",
    "    Takes in filename and outputs the header and content\n",
    "    \"\"\"\n",
    "    rdd_file = sc.textFile('data/'+ file_name +'.csv',8)\n",
    "    header, content = rid_titles(rdd_file)\n",
    "    return header, content"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "88ccfe94480345839e4fa70562c0205c",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def get_merge_items(rdd, header, key):\n",
    "    header_count = len(header)\n",
    "    key_index = header.index(key)\n",
    "    rdd_kp = rdd.map(lambda x: (x[key_index], [x[other_index] for other_index in range(header_count) if other_index!=key_index]))\n",
    "    return header_count, key_index, rdd_kp\n",
    "\n",
    "def merge_rdd(header_1, rdd_1, header_2, rdd_2, key):\n",
    "    count_1, index_1, rdd_kp_1 = get_merge_items(rdd_1, header_1, key)\n",
    "    count_2, index_2, rdd_kp_2 = get_merge_items(rdd_2, header_2, key)\n",
    "    return rdd_kp_1.join(rdd_kp_2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "86c57536230a4294abba155b28d906e7",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def spark_obj_to_df(spark_content_list, column_name_list):\n",
    "    df = pd.DataFrame.from_records(spark_content_list, columns = column_name_list)\n",
    "    return df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Prepping rdds"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "711369d34244447e8047f0a5e895aa56",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "donations_headers, donations_wo = rid_titles(donations_rdd)\n",
    "donors_headers, donors_wo = rid_titles(donors_rdd)\n",
    "small_projects_headers, small_projects_wo = rid_titles(small_projects_rdd)\n",
    "schools_headers, schools_wo = rid_titles(schools_rdd)\n",
    "teachers_headers, teachers_wo = rid_titles(teachers_rdd)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Merging"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## #1 Data Manipulation (ky)\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "merged_donations = merge_rdd(donations_headers, donations_wo, donors_headers, donors_wo, 'Donor ID' )\n",
    "\n",
    "# picking the columns that we're looking for\n",
    "concat_merge_donations = merged_donations.mapValues(lambda x: (x[0][0], x[1][1], x[0][3]))\n",
    "\n",
    "# flatten the KV rdd and creating its header\n",
    "flatten_concat_donations_headers = ['State', 'Donation Amount']\n",
    "flatten_concat_donations = concat_merge_donations.map(lambda x: (x[1][1], x[1][2]))\n",
    "\n",
    "cal_rdd = flatten_concat_donations.filter(lambda x: (x[0]=='California'))\n",
    "cal_rdd.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "ccfe954ecbc545939837eef3e7848457",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "output_folder_name = 's3a://dcgroup11/data/donor_donation.csv'\n",
    "\n",
    "# cal_rdd.saveAsTextFile(output_folder_name)  # Here will save file. check"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## #2 Data Manipulation (al)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "45af2a24236d46a0bfb19dc486718879",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "merged_donations = merge_rdd(donors_headers, donors_wo, donations_headers, donations_wo,  'Donor ID' )\n",
    "\n",
    "# manipulate\n",
    "donations_value = merged_donations.map(lambda x: (x[1][0][1], float(x[1][1][3])))\n",
    "donations_sum = donations_value.reduceByKey(lambda x,y: x+y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "879ba56486764c4da559753f505bca6c",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "output_folder_name = 's3a://dcgroup11/data/number_donors.csv'\n",
    "        \n",
    "# remove_exist(output_folder_name)  # Old folder is removed, if exists.\n",
    "\n",
    "donations_sum.saveAsTextFile(output_folder_name)  # Here will save file. check"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## #3 Data Manipulation (ktmr)\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Merging\n",
    "## #3 Data Manipulation (ktmr)\n",
    "postedDate = small_projects_wo.map(lambda x: (x[7])).map(lambda x: x if (len(x) > 0) else '2013-01-01').map(lambda x: x.split('-')[0])\n",
    "expDate = small_projects_wo.map(lambda x: (x[8])).map(lambda x: x if (len(x) > 0) else '2013-01-01').map(lambda x: x.split('-')[0])\n",
    "fundedDate = small_projects_wo.map(lambda x: (x[10])).map(lambda x: x if (len(x) > 0) else '2013-01-01').map(lambda x: x.split('-')[0])\n",
    "projectType = small_projects_wo.map(lambda x: x[4])\n",
    "category = small_projects_wo.map(lambda x: x[5])\n",
    "cost = small_projects_wo.map(lambda x: round(float(x[6])))\n",
    "status = small_projects_wo.map(lambda x: x[9])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd_names_list = ['postedDate','expDate','fundedDate','projectType','category','cost','status']\n",
    "rdd_rdd_list = [postedDate, expDate, fundedDate, projectType, category, cost, status]\n",
    "\n",
    "def save_file(title_list, rdd_list):\n",
    "    for count in len(titles):\n",
    "        output_folder_name = f's3a://dcgroup11/data/{title[count]}.csv'\n",
    "        rdd_list[count].saveAsTextFile(output_folder_name)  # Here will save file. done"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "90ce8a273a3247bbbda17277740afc9e",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# save_file(rdd_names_list, rdd_rdd_list)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## #4 Data Manipulation (ns)\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "donor_city = donors_wo.map(lambda x: x[1]).countByValue()\n",
    "\n",
    "zipper = zip(donor_city.keys(), donor_city.values())\n",
    "\n",
    "donor_city_rdd = sc.parallelize(zipper)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "7febcb1d4611477db684060c69da4495",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "output_folder_name = 's3a://dcgroup11/data/donor_city.csv'\n",
    "\n",
    "donor_city_rdd.saveAsTextFile(output_folder_name)  # Here will save file. done"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## #5 Data Manipulation (ns)\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "cefc4ada91924a24b189838be6110ddd",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "teachers_date = teachers_wo.map(lambda x: (x[2], 1)).countByKey()\n",
    "\n",
    "teachers_date_zipper = zip(teachers_date.keys(), teachers_date.values())\n",
    "\n",
    "teachers_date_rdd = sc.parallelize(teachers_date_zipper)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "878e9dc5f0d34f92a5d904469b4446a4",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "output_folder_name = 's3a://dcgroup11/data/teachers_date.csv'\n",
    "\n",
    "teachers_date_rdd.saveAsTextFile(output_folder_name)  # Here will save file. done"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## #6 Data Manipulation (al)\n",
    "---"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "bc12343591a84acc81d37a849ef7e9d9",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# gather data\n",
    "def get_month_date(obs, index):\n",
    "    # takes in observation and manipulate 1 item according to the index\n",
    "    obj = obs[index].split('-')\n",
    "    return int(obj[0]), int(obj[1])\n",
    "\n",
    "projects_dt_value = small_projects_wo.map(lambda x: (get_month_date(x, 7), float(x[6])))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "07f78722e8a24093a5c1894a6a0d731d",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "res = projects_dt_value.aggregateByKey((0,0),\n",
    "                                      lambda x, y: (x[0]+1, x[1]+y),\n",
    "                                      lambda x, y: (x[0]+y[0], x[1]+y[1]))\\\n",
    "                       .mapValues(lambda x: int(x[1]/float(x[0])))\\\n",
    "                       .sortBy(lambda x: x[1], ascending=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "b3d3069f01ba467aa1b07e2e850e913d",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# output_folder_name = 's3a://dcgroup11/data/res.csv'\n",
    "\n",
    "# res.saveAsTextFile(output_folder_name)  # Here will save file. done"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc.stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "DistributedComputing",
   "language": "python",
   "name": "distributedcomputing"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}