de-airBnb-aws.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "65c1f79e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "24dccf66",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "5982ea05",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "from io import StringIO"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "e872a1b6",
   "metadata": {},
   "outputs": [],
   "source": [
    "import configparser"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "f3ec2f49",
   "metadata": {},
   "outputs": [],
   "source": [
    "config = configparser.ConfigParser()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "b24748af",
   "metadata": {},
   "outputs": [],
   "source": [
    "config.read_file(open('cluster3.config'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "8c6d632b",
   "metadata": {},
   "outputs": [],
   "source": [
    "key = config['AWS']['KEY']\n",
    "secret = config['AWS']['SECRET']\n",
    "region = config['DWH']['REGION']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "53e46c58",
   "metadata": {},
   "outputs": [],
   "source": [
    "athenaClient = boto3.client(\n",
    "    \"athena\",\n",
    "    aws_access_key_id=key,\n",
    "    aws_secret_access_key=secret,\n",
    "    region_name=region)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "1101f0ed",
   "metadata": {},
   "outputs": [],
   "source": [
    "Dict = {}\n",
    "def download_and_load_query_results(client, query_response):\n",
    "    while True:\n",
    "        try:\n",
    "            client.get_query_results(\n",
    "                QueryExecutionId=query_response[\"QueryExecutionId\"]\n",
    "            )\n",
    "            break\n",
    "        except Exception as err:\n",
    "            if \"not yet finished\" in str(err):\n",
    "                time.sleep(2)\n",
    "            else:\n",
    "                raise err\n",
    "                \n",
    "    temp_file_location = \"athena_query_results.csv\"\n",
    "    s3_client = boto3.client(\"s3\",\n",
    "                            aws_access_key_id=AWS_ACCESS_KEY,\n",
    "                            aws_secret_access_key=AWS_SECRET_ACCESS_KEY,\n",
    "                            region_name=AWS_REGION)\n",
    "    s3_client.download_file(\n",
    "    S3_BUCKET_NAME,\n",
    "    f\"{S3_OUTPUT_DIRECTORY}/{query_response['QueryExecutionId']}.csv\",\n",
    "    temp_file_location)\n",
    "    return pd.read_csv(temp_file_location)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "42d8c240",
   "metadata": {},
   "source": [
    "### After Glue has catalogued\n",
    "\n",
    "The next steps are\n",
    "\n",
    "1) Query the Tables using Athena Client\n",
    "\n",
    "2) Establish connection to local database using psycopg2\n",
    "\n",
    "3) Move the data from athena to local data using the Jupyter NB interface\n",
    "\n",
    "4) Do the DB refactoring on JP notebook\n",
    "\n",
    "5) After completion upload the final file to S3\n",
    "\n",
    "**Challenge is the crawler is not fetching the data from the files. It is only fetching the column names. Need to check the process "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "a309f07e",
   "metadata": {},
   "outputs": [],
   "source": [
    "response = athenaClient.start_query_execution(\n",
    "    QueryString = \"SELECT * FROM enigma_jhu;\",\n",
    "    QueryExecutionContext={\"Database\":SCHEMA_NAME},\n",
    "    ResultConfiguration={\n",
    "        \"OutputLocation\":S3_STAGING_DIR,\n",
    "        \"EncryptionConfiguration\":{\"EncryptionOption\":\"SSE_S3\"}    \n",
    "    })"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "2925ece5",
   "metadata": {},
   "outputs": [],
   "source": [
    "csv_buff = StringIO()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "id": "a1668e78",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'ResponseMetadata': {'RequestId': 'DR9WZPDT7AJSRNKK',\n",
       "  'HostId': 'SmuwdPDosEwW0yMerj62GpmeBAf95ETk0TkiNqcs7dSOLc/m6BGMQI3GscyNPqmR0/Q6Rkvdma8=',\n",
       "  'HTTPStatusCode': 200,\n",
       "  'HTTPHeaders': {'x-amz-id-2': 'SmuwdPDosEwW0yMerj62GpmeBAf95ETk0TkiNqcs7dSOLc/m6BGMQI3GscyNPqmR0/Q6Rkvdma8=',\n",
       "   'x-amz-request-id': 'DR9WZPDT7AJSRNKK',\n",
       "   'date': 'Sun, 30 Oct 2022 09:53:49 GMT',\n",
       "   'etag': '\"052b91a93c128f0920be05c2066e5f13\"',\n",
       "   'server': 'AmazonS3',\n",
       "   'content-length': '0'},\n",
       "  'RetryAttempts': 0},\n",
       " 'ETag': '\"052b91a93c128f0920be05c2066e5f13\"'}"
      ]
     },
     "execution_count": 87,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "s3Reso.Object(bucket,'output/dimHospital.csv').put(Body=csv_buff.getvalue())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "id": "0d32631e",
   "metadata": {},
   "outputs": [],
   "source": [
    "dimRegion2.to_csv(csv_buff)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "f4bdb262",
   "metadata": {},
   "outputs": [],
   "source": [
    "s3 = boto3.resource('s3',\n",
    "                    region_name=\"us-east-1\",\n",
    "                    aws_access_key_id=key,\n",
    "                    aws_secret_access_key=secret)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "3f6e7127",
   "metadata": {},
   "outputs": [],
   "source": [
    "bucket = s3.Bucket(\"airbnb-fresh\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "d3eb4fa1",
   "metadata": {},
   "outputs": [],
   "source": [
    "locFile = [filename.key for filename in bucket.objects.filter(Prefix='')]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "be5533db",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['calendar.csv',\n",
       " 'listings.csv',\n",
       " 'listings_details.csv',\n",
       " 'neighbourhoods.csv',\n",
       " 'reviews.csv',\n",
       " 'reviews_details.csv']"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "locFile"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "ab47e6d8",
   "metadata": {},
   "outputs": [],
   "source": [
    "import psycopg2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "id": "d9e15950",
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    conn = psycopg2.connect(host=DWH_ENDPOINT,dbname=DB_NAME,user=DB_USER,password=DWH_DB_PASSWORD,port=DWH_PORT)\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 114,
   "id": "cce5b260",
   "metadata": {},
   "outputs": [],
   "source": [
    "conn.set_session(autocommit=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "id": "9dffd99d",
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    cur = conn.cursor()\n",
    "    \n",
    "except:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 133,
   "id": "4da312ec",
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    cur.execute(\"\"\"\n",
    "        CREATE TABLE \"factcorrected\" (\n",
    "            \"index\" INTEGER,\n",
    "          \"fips\" REAL,\n",
    "          \"province_state\" TEXT,\n",
    "          \"country_region\" TEXT,\n",
    "          \"confirmed\" REAL,\n",
    "          \"deaths\" REAL,\n",
    "          \"recovered\" REAL,\n",
    "          \"active\" REAL,\n",
    "          \"date\" INTEGER,\n",
    "          \"state\" TEXT,\n",
    "          \"positive\" REAL,\n",
    "          \"negative\" REAL,\n",
    "          \"hospitalizedcurrently\" REAL,\n",
    "          \"hospitalized\" REAL,\n",
    "          \"hospitalizeddischarged\" REAL\n",
    "    )\n",
    "    \"\"\")\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 125,
   "id": "a081974e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'arn:aws:iam::642924624251:role/redshiftRole'"
      ]
     },
     "execution_count": 125,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "DWH_ROLE_ARN"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 134,
   "id": "4cc47445",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Load into table 'factcorrected' failed.  Check 'stl_load_errors' system table for details.\n",
      "\n"
     ]
    }
   ],
   "source": [
    "try:\n",
    "    cur.execute(\"\"\"\n",
    "        COPY factcorrected from 's3://de-glued/output/factcovid.csv'\n",
    "        CREDENTIALS 'aws_iam_role=arn:aws:iam::642924624251:role/redshiftRole'\n",
    "        DELIMITER ','\n",
    "        region 'us-east-1'\n",
    "    \"\"\")\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "016b2cc4",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.10.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}