4.validate_completed_data.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Validate Completed Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "sys.path.append('airflow/dags/lib')\n",
    "import emrspark_lib as emrs\n",
    "import configparser\n",
    "import time\n",
    "\n",
    "import logging\n",
    "import os\n",
    "import json\n",
    "\n",
    "logger = logging.getLogger()\n",
    "logger.setLevel(logging.INFO)\n",
    "\n",
    "config = configparser.ConfigParser()\n",
    "config.read('airflow/config.cfg')\n",
    "\n",
    "CLUSTER_NAME = config['AWS']['CLUSTER_NAME']\n",
    "VPC_ID = config['AWS']['VPC_ID']\n",
    "SUBNET_ID = config['AWS']['SUBNET_ID']\n",
    "\n",
    "if config['App']['STOCKS'] == '':\n",
    "    STOCKS = []\n",
    "else:\n",
    "    STOCKS = json.loads(config.get('App', 'STOCKS').replace(\"'\", '\"'))\n",
    "\n",
    "ec2, emr, iam = emrs.get_boto_clients(config['AWS']['REGION_NAME'], config=config)\n",
    "\n",
    "if VPC_ID == '':\n",
    "    VPC_ID = emrs.get_first_available_vpc(ec2)\n",
    "\n",
    "if SUBNET_ID == '':\n",
    "    SUBNET_ID = emrs.get_first_available_subnet(ec2, VPC_ID)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "master_sg_id = emrs.create_security_group(ec2, '{}SG'.format(CLUSTER_NAME),\n",
    "    'Master SG for {}'.format(CLUSTER_NAME), VPC_ID)\n",
    "slave_sg_id = emrs.create_security_group(ec2, '{}SlaveSG'.format(CLUSTER_NAME),\n",
    "    'Slave SG for {}'.format(CLUSTER_NAME), VPC_ID)\n",
    "\n",
    "keypair = emrs.create_key_pair(ec2, '{}_pem'.format(CLUSTER_NAME))\n",
    "\n",
    "emrs.create_default_roles(iam)\n",
    "emrs.wait_for_roles(iam)\n",
    "\n",
    "cluster_id = emrs.create_emr_cluster(emr, CLUSTER_NAME,\n",
    "                master_sg_id,\n",
    "                slave_sg_id,\n",
    "                keypair['KeyName'], SUBNET_ID,\n",
    "                release_label='emr-5.28.1',\n",
    "                num_core_nodes=3)\n",
    "cluster_dns = emrs.get_cluster_dns(emr, cluster_id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "args_v = {\n",
    "            'PULL_DATE': '2020-12-10',\n",
    "            'AWS_ACCESS_KEY_ID': config['AWS']['AWS_ACCESS_KEY_ID'],\n",
    "            'AWS_SECRET_ACCESS_KEY': config['AWS']['AWS_SECRET_ACCESS_KEY'],\n",
    "            'DB_HOST': config['App']['DB_HOST'],\n",
    "            'TABLE_SHORT_ANALYSIS': config['App']['TABLE_SHORT_ANALYSIS_QUANTOPIAN'],\n",
    "            'TABLE_SHORT_INTERESTS_NASDAQ': config['App']['TABLE_SHORT_INTERESTS_NASDAQ']\n",
    "        }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "emrs.kill_all_spark_sessions(cluster_dns)\n",
    "session_headers = emrs.create_spark_session(cluster_dns)\n",
    "emrs.wait_for_spark(cluster_dns, session_headers)\n",
    "job_response_headers = emrs.submit_spark_job_from_file(\n",
    "        cluster_dns, session_headers,\n",
    "        # Any debugging/validate_* can be included here. Don't forget to update the `args_v` dictionary above.\n",
    "        'debugging/validate_raw_short_interests.py',\n",
    "        args=args_v,\n",
    "        commonpath='airflow/dags/etl/common.py',\n",
    "        helperspath='airflow/dags/etl/helpers.py'\n",
    ")\n",
    "final_status, logs = emrs.track_spark_job(cluster_dns, job_response_headers, sleep_seconds=10)\n",
    "emrs.kill_spark_session(cluster_dns, session_headers)\n",
    "for line in logs:\n",
    "    logging.info(line)\n",
    "    if '(FAIL)' in str(line):\n",
    "        logging.error(line)\n",
    "        raise Exception(\"ETL process fails.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "emrs.delete_cluster(emr, cluster_id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "df = pd.DataFrame(data=[{'Symbol': 'X', 'last_date': '2020-01-01'}, {'Symbol': 'Y', 'last_date': '2020-01-01'}])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.loc[df['Symbol'] == 'z']['last_date'].values"
   ]
  }
 ],
 "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.7.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}