notebook.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import boto3\n",
    "import json\n",
    "import configparser"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "config = configparser.ConfigParser()\n",
    "config.read_file(open('dwh.cfg'))\n",
    "KEY                    = config.get('AWS','KEY')\n",
    "SECRET                 = config.get('AWS','SECRET')\n",
    "DWH_IAM_ROLE_NAME      = config.get('CLUSTER', 'DWH_IAM_ROLE_NAME')\n",
    "\n",
    "DWH_CLUSTER_TYPE       = config.get('CLUSTER', 'DWH_CLUSTER_TYPE')\n",
    "DWH_NUM_NODES          = config.get('CLUSTER', 'DWH_NUM_NODES')\n",
    "DWH_NODE_TYPE          = config.get('CLUSTER', 'DWH_NODE_TYPE')\n",
    "\n",
    "DWH_CLUSTER_IDENTIFIER = config.get('CLUSTER', 'DWH_CLUSTER_IDENTIFIER')\n",
    "DB_NAME                = config.get('CLUSTER', 'DB_NAME')\n",
    "DB_USER                = config.get('CLUSTER', 'DB_USER')\n",
    "DB_PASSWORD            = config.get('CLUSTER', 'DB_PASSWORD')\n",
    "DB_PORT                = config.get('CLUSTER', 'DB_PORT')\n",
    "ARN                = config.get('IAM_ROLE', 'ARN')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "ec2 = boto3.resource('ec2',\n",
    "                       region_name=\"us-west-2\",\n",
    "                       aws_access_key_id=KEY,\n",
    "                       aws_secret_access_key=SECRET\n",
    "                    )\n",
    "\n",
    "s3 = boto3.resource('s3',\n",
    "                       region_name=\"us-west-2\",\n",
    "                       aws_access_key_id=KEY,\n",
    "                       aws_secret_access_key=SECRET\n",
    "                   )\n",
    "\n",
    "iam = boto3.client('iam',aws_access_key_id=KEY,\n",
    "                     aws_secret_access_key=SECRET,\n",
    "                       region_name=\"us-west-2\"\n",
    "                  )\n",
    "\n",
    "redshift = boto3.client('redshift',\n",
    "                       region_name=\"us-west-2\",\n",
    "                       aws_access_key_id=KEY,\n",
    "                       aws_secret_access_key=SECRET\n",
    "                       )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1.1 Creating a new IAM Role\n",
      "An error occurred (AccessDenied) when calling the CreateRole operation: User: arn:aws:iam::783882260760:user/airflow_redshift_user is not authorized to perform: iam:CreateRole on resource: arn:aws:iam::783882260760:role/AWSServiceRoleForRedshift\n",
      "1.2 Attaching Policy\n"
     ]
    },
    {
     "ename": "ClientError",
     "evalue": "An error occurred (AccessDenied) when calling the AttachRolePolicy operation: User: arn:aws:iam::783882260760:user/airflow_redshift_user is not authorized to perform: iam:AttachRolePolicy on resource: role AWSServiceRoleForRedshift",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mClientError\u001b[0m                               Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-16-3596c08f35af>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m()\u001b[0m\n\u001b[1;32m     21\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     22\u001b[0m iam.attach_role_policy(RoleName=DWH_IAM_ROLE_NAME,\n\u001b[0;32m---> 23\u001b[0;31m                        \u001b[0mPolicyArn\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m\"arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess\"\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     24\u001b[0m                       )['ResponseMetadata']['HTTPStatusCode']\n\u001b[1;32m     25\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/opt/conda/lib/python3.6/site-packages/botocore/client.py\u001b[0m in \u001b[0;36m_api_call\u001b[0;34m(self, *args, **kwargs)\u001b[0m\n\u001b[1;32m    318\u001b[0m                     \"%s() only accepts keyword arguments.\" % py_operation_name)\n\u001b[1;32m    319\u001b[0m             \u001b[0;31m# The \"self\" in this scope is referring to the BaseClient.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 320\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_make_api_call\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0moperation_name\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mkwargs\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    321\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    322\u001b[0m         \u001b[0m_api_call\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m__name__\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mstr\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mpy_operation_name\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/opt/conda/lib/python3.6/site-packages/botocore/client.py\u001b[0m in \u001b[0;36m_make_api_call\u001b[0;34m(self, operation_name, api_params)\u001b[0m\n\u001b[1;32m    621\u001b[0m             \u001b[0merror_code\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mparsed_response\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mget\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Error\"\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m{\u001b[0m\u001b[0;34m}\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mget\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Code\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    622\u001b[0m             \u001b[0merror_class\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mexceptions\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mfrom_code\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0merror_code\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 623\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0merror_class\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mparsed_response\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0moperation_name\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    624\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    625\u001b[0m             \u001b[0;32mreturn\u001b[0m \u001b[0mparsed_response\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mClientError\u001b[0m: An error occurred (AccessDenied) when calling the AttachRolePolicy operation: User: arn:aws:iam::783882260760:user/airflow_redshift_user is not authorized to perform: iam:AttachRolePolicy on resource: role AWSServiceRoleForRedshift"
     ]
    }
   ],
   "source": [
    "from botocore.exceptions import ClientError\n",
    "\n",
    "#1.1 Create the role, \n",
    "try:\n",
    "    print(\"1.1 Creating a new IAM Role\") \n",
    "    dwhRole = iam.create_role(\n",
    "        Path='/',\n",
    "        RoleName=DWH_IAM_ROLE_NAME,\n",
    "        Description = \"Allows Redshift clusters to call AWS services on your behalf.\",\n",
    "        AssumeRolePolicyDocument=json.dumps(\n",
    "            {'Statement': [{'Action': 'sts:AssumeRole',\n",
    "               'Effect': 'Allow',\n",
    "               'Principal': {'Service': 'redshift.amazonaws.com'}}],\n",
    "             'Version': '2012-10-17'})\n",
    "    )    \n",
    "except Exception as e:\n",
    "    print(e)\n",
    "    \n",
    "    \n",
    "print(\"1.2 Attaching Policy\")\n",
    "\n",
    "iam.attach_role_policy(RoleName=DWH_IAM_ROLE_NAME,\n",
    "                       PolicyArn=\"arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess\"\n",
    "                      )['ResponseMetadata']['HTTPStatusCode']\n",
    "\n",
    "print(\"1.3 Get the IAM role ARN\")\n",
    "roleArn = iam.get_role(RoleName=DWH_IAM_ROLE_NAME)['Role']['Arn']\n",
    "\n",
    "print(roleArn)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "An error occurred (ClusterAlreadyExists) when calling the CreateCluster operation: Cluster already exists\n"
     ]
    }
   ],
   "source": [
    "try:\n",
    "    response = redshift.create_cluster(        \n",
    "        #HW\n",
    "        ClusterType=DWH_CLUSTER_TYPE,\n",
    "        NodeType=DWH_NODE_TYPE,\n",
    "        NumberOfNodes=int(DWH_NUM_NODES),\n",
    "\n",
    "        #Identifiers & Credentials\n",
    "        DBName=DB_NAME,\n",
    "        ClusterIdentifier=DWH_CLUSTER_IDENTIFIER,\n",
    "        MasterUsername=DB_USER,\n",
    "        MasterUserPassword=DB_PASSWORD,\n",
    "        \n",
    "        #Roles (for s3 access)\n",
    "        IamRoles=[roleArn]  \n",
    "    )\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Key</th>\n",
       "      <th>Value</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ClusterIdentifier</td>\n",
       "      <td>dwhcluster</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>NodeType</td>\n",
       "      <td>dc2.large</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>ClusterStatus</td>\n",
       "      <td>available</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>MasterUsername</td>\n",
       "      <td>idelfonso</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>DBName</td>\n",
       "      <td>sparkifydb</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>Endpoint</td>\n",
       "      <td>{'Address': 'dwhcluster.cnaqoizhiv4v.us-west-2.redshift.amazonaws.com', 'Port': 5439}</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>VpcId</td>\n",
       "      <td>vpc-b2a48cd7</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>NumberOfNodes</td>\n",
       "      <td>4</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                 Key  \\\n",
       "0  ClusterIdentifier   \n",
       "1  NodeType            \n",
       "2  ClusterStatus       \n",
       "3  MasterUsername      \n",
       "4  DBName              \n",
       "5  Endpoint            \n",
       "6  VpcId               \n",
       "7  NumberOfNodes       \n",
       "\n",
       "                                                                                   Value  \n",
       "0  dwhcluster                                                                             \n",
       "1  dc2.large                                                                              \n",
       "2  available                                                                              \n",
       "3  idelfonso                                                                              \n",
       "4  sparkifydb                                                                             \n",
       "5  {'Address': 'dwhcluster.cnaqoizhiv4v.us-west-2.redshift.amazonaws.com', 'Port': 5439}  \n",
       "6  vpc-b2a48cd7                                                                           \n",
       "7  4                                                                                      "
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "def prettyRedshiftProps(props):\n",
    "    pd.set_option('display.max_colwidth', -1)\n",
    "    keysToShow = [\"ClusterIdentifier\", \"NodeType\", \"ClusterStatus\", \"MasterUsername\", \"DBName\", \"Endpoint\", \"NumberOfNodes\", 'VpcId']\n",
    "    x = [(k, v) for k,v in props.items() if k in keysToShow]\n",
    "    return pd.DataFrame(data=x, columns=[\"Key\", \"Value\"])\n",
    "\n",
    "myClusterProps = redshift.describe_clusters(ClusterIdentifier=DWH_CLUSTER_IDENTIFIER)['Clusters'][0]\n",
    "prettyRedshiftProps(myClusterProps)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "DWH_ENDPOINT ::  dwhcluster.cnaqoizhiv4v.us-west-2.redshift.amazonaws.com\n",
      "DWH_ROLE_ARN ::  arn:aws:iam::783882260760:role/dwhRole\n"
     ]
    }
   ],
   "source": [
    "DWH_ENDPOINT = myClusterProps['Endpoint']['Address']\n",
    "DWH_ROLE_ARN = myClusterProps['IamRoles'][0]['IamRoleArn']\n",
    "print(\"DWH_ENDPOINT :: \", DWH_ENDPOINT)\n",
    "print(\"DWH_ROLE_ARN :: \", DWH_ROLE_ARN)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "An error occurred (InvalidPermission.Duplicate) when calling the AuthorizeSecurityGroupIngress operation: the specified rule \"peer: 0.0.0.0/0, TCP, from port: 5439, to port: 5439, ALLOW\" already exists\n"
     ]
    }
   ],
   "source": [
    "try:\n",
    "    vpc = ec2.Vpc(id=myClusterProps['VpcId'])\n",
    "    defaultSg = list(vpc.security_groups.all())[0]\n",
    "\n",
    "    defaultSg.authorize_ingress(\n",
    "        GroupName=defaultSg.group_name,\n",
    "        CidrIp='0.0.0.0/0',\n",
    "        IpProtocol='TCP',\n",
    "        FromPort=int(DB_PORT),\n",
    "        ToPort=int(DB_PORT))\n",
    "except Exception as e:\n",
    "    print(e)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "# Hay que limpiar las vainas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'ResponseMetadata': {'RequestId': 'd17000e5-e7d6-4237-8331-adba620db8ae',\n",
       "  'HTTPStatusCode': 200,\n",
       "  'HTTPHeaders': {'x-amzn-requestid': 'd17000e5-e7d6-4237-8331-adba620db8ae',\n",
       "   'content-type': 'text/xml',\n",
       "   'content-length': '200',\n",
       "   'date': 'Sun, 19 Jul 2020 20:39:48 GMT'},\n",
       "  'RetryAttempts': 0}}"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "redshift.delete_cluster( ClusterIdentifier=DWH_CLUSTER_IDENTIFIER,  SkipFinalClusterSnapshot=True)\n",
    "iam.detach_role_policy(RoleName=DWH_IAM_ROLE_NAME, PolicyArn=\"arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess\")\n",
    "iam.delete_role(RoleName=DWH_IAM_ROLE_NAME)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "editable": true
   },
   "source": [
    "# Creando una conexcion para ejecutar entradas a mi base de datos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": [
    "import os \n",
    "import configparser\n",
    "import psycopg2\n",
    "from sql_queries import create_table_queries, drop_table_queries\n",
    "\n",
    "conn_string=\"postgresql://{}:{}@{}:{}/{}\".format(*config['CLUSTER'].values())\n",
    "conn = psycopg2.connect(conn_string)\n",
    "cur = conn.cursor()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Connected: idelfonso@sparkifydb'"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%load_ext sql\n",
    "%sql $conn_string"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " * postgresql://idelfonso:***@dwhcluster.cnaqoizhiv4v.us-west-2.redshift.amazonaws.com:5439/sparkifydb\n",
      "0 rows affected.\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "    <tr>\n",
       "        <th>num_songs</th>\n",
       "        <th>artist_id</th>\n",
       "        <th>artist_latitude</th>\n",
       "        <th>artist_longitude</th>\n",
       "        <th>artitst_location</th>\n",
       "        <th>artitst_name</th>\n",
       "        <th>song_id</th>\n",
       "        <th>title</th>\n",
       "        <th>duration</th>\n",
       "        <th>year</th>\n",
       "    </tr>\n",
       "</table>"
      ],
      "text/plain": [
       "[]"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%sql\n",
    "SELECT *\n",
    "FROM staging_songs\n",
    "limit 4;"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {
    "editable": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " * postgresql://idelfonso:***@dwhcluster.cnaqoizhiv4v.us-west-2.redshift.amazonaws.com:5439/sparkifydb\n",
      "   postgresql://idelfonso:***@dwhcluster.cxna1iqaysmu.us-east-2.redshift.amazonaws.com:5439/sparkifydb\n",
      "0 rows affected.\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "    <tr>\n",
       "        <th>userid</th>\n",
       "        <th>slice</th>\n",
       "        <th>tbl</th>\n",
       "        <th>starttime</th>\n",
       "        <th>session</th>\n",
       "        <th>query</th>\n",
       "        <th>filename</th>\n",
       "        <th>line_number</th>\n",
       "        <th>colname</th>\n",
       "        <th>type</th>\n",
       "        <th>col_length</th>\n",
       "        <th>position</th>\n",
       "        <th>raw_line</th>\n",
       "        <th>raw_field_value</th>\n",
       "        <th>err_code</th>\n",
       "        <th>err_reason</th>\n",
       "    </tr>\n",
       "</table>"
      ],
      "text/plain": [
       "[]"
      ]
     },
     "execution_count": 82,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%sql select * from stl_load_errors;"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "editable": true
   },
   "outputs": [],
   "source": []
  },
  {
   "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
}