workspace/data-engineering/06-immigration-capstone/etl-s3-testing.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "### ETL notebook for testing the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "import configparser\n",
    "import pandas as pd\n",
    "from datetime import datetime\n",
    "from pyspark.sql.functions import dayofweek\n",
    "import os\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import udf, col, lit\n",
    "from pyspark.sql.functions import year, month, dayofmonth, hour, weekofyear, date_format, to_date, upper\n",
    "import logging\n",
    "from pyspark.sql.types import DateType\n",
    "from pyspark.sql.functions import monotonically_increasing_id"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "# CONFIG\n",
    "config = configparser.ConfigParser()\n",
    "config.read('dl.cfg')\n",
    "\n",
    "KEY = config.get('AWS', 'AWS_ACCESS_KEY_ID')\n",
    "SECRET = config.get('AWS', 'AWS_SECRET_ACCESS_KEY')\n",
    "output_data = config.get('S3', 'DEST_S3_BUCKET')\n",
    "\n",
    "\n",
    "os.environ['AWS_ACCESS_KEY_ID']=KEY\n",
    "os.environ['AWS_SECRET_ACCESS_KEY']=SECRET\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "def create_spark_session():\n",
    "    spark = SparkSession \\\n",
    "        .builder \\\n",
    "        .config(\"spark.jars.repositories\", \"https://repos.spark-packages.org/\")\\\n",
    "        .config(\"spark.jars.packages\", \"org.apache.hadoop:hadoop-aws:2.7.0,saurfang:spark-sas7bdat:2.0.0-s_2.11\")\\\n",
    "        .enableHiveSupport().getOrCreate()\n",
    "    return spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "spark = create_spark_session()\n",
    "input_data ='./'\n",
    "output_data = 's3a://gfp-udacity/testing'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "def rename_columns(table, new_columns):\n",
    "    for original, new in zip(table.columns, new_columns):\n",
    "        table = table.withColumnRenamed(original, new)\n",
    "    return table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "def process_demography_data(spark, input_data, output_data):\n",
    "    \"\"\" Process demograpy data to get dim_demog_population \n",
    "     and d_demog_statistics table\n",
    "        Arguments:\n",
    "            spark {object}: SparkSession object\n",
    "            input_data {object}: Source S3 endpoint\n",
    "            output_data {object}: Target S3 endpoint\n",
    "        Returns:\n",
    "            None\n",
    "    \"\"\"\n",
    "\n",
    "    logging.info(\"Start processing d_demog_statistics\")\n",
    "    # read demography data file\n",
    "    demog_data = os.path.join(input_data + 'us-cities-demographics.csv')\n",
    "    df = spark.read.format('csv').options(header=True, delimiter=';').load(demog_data)\n",
    "\n",
    "\n",
    "    d_demog_statistics = df.select(['City', 'State', 'Male Population', 'Female Population', \\\n",
    "                              'Number of Veterans', 'Foreign-born', 'Race']).distinct() \\\n",
    "                              .withColumn(\"demog_pop_id\", monotonically_increasing_id())\n",
    "\n",
    "\n",
    "    new_columns = ['city', 'state', 'male_population', 'female_population', \\\n",
    "                   'num_vetarans', 'foreign_born', 'race']\n",
    "    d_demog_statistics = rename_columns(d_demog_statistics, new_columns)\n",
    "\n",
    "    # write dim_demog_population table to parquet files\n",
    "    d_demog_statistics.write.mode(\"overwrite\")\\\n",
    "                        .parquet(path=output_data + 'd_demog_statistics')\n",
    "\n",
    "    \n",
    "    logging.info(\"Start processing d_demog_statistics\")\n",
    "    d_demog_statistics = df.select(['City', 'State', 'Median Age', 'Average Household Size'])\\\n",
    "                             .distinct()\\\n",
    "                             .withColumn(\"d_demog_statistics\", monotonically_increasing_id())\n",
    "\n",
    "    new_columns = ['city', 'state', 'median_age', 'avg_household_size']\n",
    "    d_demog_statistics = rename_columns(d_demog_statistics, new_columns)\n",
    "    d_demog_statistics = d_demog_statistics.withColumn('city', upper(col('city')))\n",
    "    d_demog_statistics = d_demog_statistics.withColumn('state', upper(col('state')))\n",
    "\n",
    "    # write dim_demog_statistics table to parquet files\n",
    "    d_demog_statistics.write.mode(\"overwrite\")\\\n",
    "                        .parquet(path=output_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "process_demography_data(spark, input_data, output_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}