7_Deploy_Model_Glue.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Name\n",
    "ModelName = 'titanic_propensity_survive'\n",
    "# ATENÇÃO: nome do bucket criado no S3 (altere para o bucket com o seu nome)\n",
    "bucketName = 'aula-deploy-modelos-seu-nome'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    import sys\n",
    "    sys.path.append('/var/lang/lib/python37.zip')\n",
    "    sys.path.append('/var/lang/lib/python3.7')\n",
    "    sys.path.append('/var/lang/lib/python3.7/lib-dynload')\n",
    "    sys.path.append('/var/lang/lib/python3.7/site-packages')\n",
    "    sys.path.remove('/opt/.sagemakerinternal/conda/lib/python3.7/site-packages')\n",
    "except:\n",
    "    pass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "import json\n",
    "import awswrangler as wr\n",
    "from decimal import Decimal\n",
    "import time\n",
    "import findspark\n",
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.context import SparkContext\n",
    "from pyspark.sql.functions import *\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.types import *\n",
    "import awswrangler as wr\n",
    "from pysparkling.ml import H2OMOJOSettings\n",
    "from pysparkling.ml import H2OMOJOModel\n",
    "from sagemaker import get_execution_role"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "my_boto3_session = boto3.Session(region_name='us-east-1')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Criar o contexto Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<module 'pyspark' from '/var/lang/lib/python3.7/site-packages/pyspark/__init__.py'>\n"
     ]
    }
   ],
   "source": [
    "print(pyspark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "22/10/21 20:20:48 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "22/10/21 20:20:49 WARN SparkContext: The jar file:///var/lang/lib/python3.7/site-packages/sparkling_water/sparkling_water_assembly.jar has been added already. Overwriting of added jars is not supported in the current version.\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://aula-deploy-modelos-ml-t3-medium-5a0e1fd7fcbb014cb643c0daa31f:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.1.1</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>SparkContext</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7f8e3acfbf50>"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Confirmar se o camimho abaixo estão igual ao print acima. Se não estiver altere para o valor do print\n",
    "findspark.init('/var/lang/lib/python3.7/site-packages/pyspark')\n",
    "sc = SparkContext('local', 'SparkContext')\n",
    "spark = SparkSession.builder.appName(\"SparkSession\").getOrCreate()\n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Criar um dataframe Spark usando os dados da tabela auladeploymodelos.titanic_propensity_survive"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "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>pclass</th>\n",
       "      <th>embarked</th>\n",
       "      <th>cabine_prefix</th>\n",
       "      <th>ticket_str</th>\n",
       "      <th>nametitle</th>\n",
       "      <th>fare</th>\n",
       "      <th>sibsp</th>\n",
       "      <th>parch</th>\n",
       "      <th>age_mean</th>\n",
       "      <th>ticket_int</th>\n",
       "      <th>survived</th>\n",
       "      <th>passengerid</th>\n",
       "      <th>referencedate</th>\n",
       "      <th>partition_0</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>3</td>\n",
       "      <td>S</td>\n",
       "      <td>missing</td>\n",
       "      <td>A</td>\n",
       "      <td>Mr</td>\n",
       "      <td>7.2500</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>22.000000</td>\n",
       "      <td>521171.0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>1912-05</td>\n",
       "      <td>train_data</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>1</td>\n",
       "      <td>C</td>\n",
       "      <td>C</td>\n",
       "      <td>PC</td>\n",
       "      <td>Mrs</td>\n",
       "      <td>71.2833</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>38.000000</td>\n",
       "      <td>17599.0</td>\n",
       "      <td>1</td>\n",
       "      <td>2</td>\n",
       "      <td>1912-04</td>\n",
       "      <td>train_data</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>3</td>\n",
       "      <td>S</td>\n",
       "      <td>missing</td>\n",
       "      <td>STONO</td>\n",
       "      <td>Miss</td>\n",
       "      <td>7.9250</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>26.000000</td>\n",
       "      <td>23101282.0</td>\n",
       "      <td>1</td>\n",
       "      <td>3</td>\n",
       "      <td>1912-06</td>\n",
       "      <td>train_data</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>1</td>\n",
       "      <td>S</td>\n",
       "      <td>C</td>\n",
       "      <td>missing</td>\n",
       "      <td>Mrs</td>\n",
       "      <td>53.1000</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>35.000000</td>\n",
       "      <td>113803.0</td>\n",
       "      <td>1</td>\n",
       "      <td>4</td>\n",
       "      <td>1912-04</td>\n",
       "      <td>train_data</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>3</td>\n",
       "      <td>S</td>\n",
       "      <td>missing</td>\n",
       "      <td>missing</td>\n",
       "      <td>Mrs</td>\n",
       "      <td>11.1333</td>\n",
       "      <td>0.0</td>\n",
       "      <td>2.0</td>\n",
       "      <td>27.000000</td>\n",
       "      <td>347742.0</td>\n",
       "      <td>1</td>\n",
       "      <td>9</td>\n",
       "      <td>1912-05</td>\n",
       "      <td>train_data</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>886</th>\n",
       "      <td>3</td>\n",
       "      <td>C</td>\n",
       "      <td>missing</td>\n",
       "      <td>missing</td>\n",
       "      <td>Miss</td>\n",
       "      <td>7.2250</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>15.000000</td>\n",
       "      <td>2667.0</td>\n",
       "      <td>1</td>\n",
       "      <td>876</td>\n",
       "      <td>1912-07</td>\n",
       "      <td>test_data</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>887</th>\n",
       "      <td>3</td>\n",
       "      <td>S</td>\n",
       "      <td>missing</td>\n",
       "      <td>missing</td>\n",
       "      <td>Mr</td>\n",
       "      <td>9.8458</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>20.000000</td>\n",
       "      <td>7534.0</td>\n",
       "      <td>0</td>\n",
       "      <td>877</td>\n",
       "      <td>1912-07</td>\n",
       "      <td>test_data</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>888</th>\n",
       "      <td>3</td>\n",
       "      <td>S</td>\n",
       "      <td>missing</td>\n",
       "      <td>missing</td>\n",
       "      <td>Mr</td>\n",
       "      <td>7.8958</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>29.699118</td>\n",
       "      <td>349217.0</td>\n",
       "      <td>0</td>\n",
       "      <td>879</td>\n",
       "      <td>1912-07</td>\n",
       "      <td>test_data</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>889</th>\n",
       "      <td>3</td>\n",
       "      <td>S</td>\n",
       "      <td>missing</td>\n",
       "      <td>missing</td>\n",
       "      <td>Mr</td>\n",
       "      <td>7.8958</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>33.000000</td>\n",
       "      <td>349257.0</td>\n",
       "      <td>0</td>\n",
       "      <td>882</td>\n",
       "      <td>1912-07</td>\n",
       "      <td>test_data</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>890</th>\n",
       "      <td>1</td>\n",
       "      <td>C</td>\n",
       "      <td>C</td>\n",
       "      <td>missing</td>\n",
       "      <td>Mr</td>\n",
       "      <td>30.0000</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>26.000000</td>\n",
       "      <td>111369.0</td>\n",
       "      <td>1</td>\n",
       "      <td>890</td>\n",
       "      <td>1912-07</td>\n",
       "      <td>test_data</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>891 rows × 14 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "     pclass embarked cabine_prefix ticket_str nametitle     fare  sibsp  \\\n",
       "0         3        S       missing          A        Mr   7.2500    1.0   \n",
       "1         1        C             C         PC       Mrs  71.2833    1.0   \n",
       "2         3        S       missing      STONO      Miss   7.9250    0.0   \n",
       "3         1        S             C    missing       Mrs  53.1000    1.0   \n",
       "4         3        S       missing    missing       Mrs  11.1333    0.0   \n",
       "..      ...      ...           ...        ...       ...      ...    ...   \n",
       "886       3        C       missing    missing      Miss   7.2250    0.0   \n",
       "887       3        S       missing    missing        Mr   9.8458    0.0   \n",
       "888       3        S       missing    missing        Mr   7.8958    0.0   \n",
       "889       3        S       missing    missing        Mr   7.8958    0.0   \n",
       "890       1        C             C    missing        Mr  30.0000    0.0   \n",
       "\n",
       "     parch   age_mean  ticket_int  survived  passengerid referencedate  \\\n",
       "0      0.0  22.000000    521171.0         0            1       1912-05   \n",
       "1      0.0  38.000000     17599.0         1            2       1912-04   \n",
       "2      0.0  26.000000  23101282.0         1            3       1912-06   \n",
       "3      0.0  35.000000    113803.0         1            4       1912-04   \n",
       "4      2.0  27.000000    347742.0         1            9       1912-05   \n",
       "..     ...        ...         ...       ...          ...           ...   \n",
       "886    0.0  15.000000      2667.0         1          876       1912-07   \n",
       "887    0.0  20.000000      7534.0         0          877       1912-07   \n",
       "888    0.0  29.699118    349217.0         0          879       1912-07   \n",
       "889    0.0  33.000000    349257.0         0          882       1912-07   \n",
       "890    0.0  26.000000    111369.0         1          890       1912-07   \n",
       "\n",
       "    partition_0  \n",
       "0    train_data  \n",
       "1    train_data  \n",
       "2    train_data  \n",
       "3    train_data  \n",
       "4    train_data  \n",
       "..          ...  \n",
       "886   test_data  \n",
       "887   test_data  \n",
       "888   test_data  \n",
       "889   test_data  \n",
       "890   test_data  \n",
       "\n",
       "[891 rows x 14 columns]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Primeiro vamos criar um dataframe pandas executando uma query no Athena\n",
    "query = \"SELECT * FROM auladeploymodelos.titanic_propensity_survive;\"\n",
    "dataprep_df = wr.athena.read_sql_query(query, database=\"auladeploymodelos\", boto3_session=my_boto3_session)\n",
    "dataprep_df = dataprep_df.apply(lambda x: x.fillna(0) if x.dtype.kind in 'biufc' else x.fillna('missing'))\n",
    "dataprep_df = dataprep_df.fillna(0)\n",
    "dataprep_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- pclass: long (nullable = true)\n",
      " |-- embarked: string (nullable = true)\n",
      " |-- cabine_prefix: string (nullable = true)\n",
      " |-- ticket_str: string (nullable = true)\n",
      " |-- nametitle: string (nullable = true)\n",
      " |-- fare: double (nullable = true)\n",
      " |-- sibsp: double (nullable = true)\n",
      " |-- parch: double (nullable = true)\n",
      " |-- age_mean: double (nullable = true)\n",
      " |-- ticket_int: double (nullable = true)\n",
      " |-- survived: long (nullable = true)\n",
      " |-- passengerid: long (nullable = true)\n",
      " |-- referencedate: string (nullable = true)\n",
      " |-- partition_0: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Depois vamos criar um dataframe Spark usando os dados do dataframe\n",
    "datasource_titanic = spark.createDataFrame(dataprep_df)\n",
    "datasource_titanic.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Criar o código que usaremos para criar um deploy usando o Glue Jobs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Após testar esse código, colocar no bloco abaixo, na parte reservada para o seu scritp de deploy\n",
    "\n",
    "#ModelPath\n",
    "PathModelMojo='./output_model/models/best/StackedEnsemble_BestOfFamily_4_AutoML_1_20221011_230015.zip'\n",
    "\n",
    "## Selected Features: variaveis que entraram no modelo\n",
    "CAT = [\n",
    "'pclass'\n",
    ",'embarked'\n",
    ",'cabine_prefix'\n",
    ",'ticket_str'\n",
    ",'nametitle']\n",
    "\n",
    "#float\n",
    "NUM = [\n",
    "'fare'\n",
    ",'sibsp'\n",
    ",'parch'\n",
    ",'age_mean'\n",
    ",'ticket_int']\n",
    "selected_features = CAT + NUM\n",
    "\n",
    "##functions\n",
    "def ratings(p1):\n",
    "    if p1 <= 0.2508362656036639:\n",
    "        return 1\n",
    "    elif p1 <= 0.6540492277407066:\n",
    "        return 2\n",
    "    else:\n",
    "        return 3\n",
    "ratings_func = F.udf(ratings, StringType())\n",
    "\n",
    "def predict_survive(predict):\n",
    "    if predict == 0:\n",
    "        return 'Not survive'\n",
    "    elif predict == 1:\n",
    "        return 'Survive'\n",
    "    else:\n",
    "        return 'predict_ERROR'\n",
    "predict_func = F.udf(predict_survive, StringType())\n",
    "\n",
    "## Escorar a base, criar o score e o rating\n",
    "settings = H2OMOJOSettings(convertUnknownCategoricalLevelsToNa = True, convertInvalidNumbersToNa = True)\n",
    "model = H2OMOJOModel.createFromMojo(PathModelMojo, settings)\n",
    "predict_titanic_1 = model.transform(datasource_titanic)\n",
    "\n",
    "predict_titanic = predict_titanic_1.withColumn('predict_int', F.col('detailed_prediction.label').cast(IntegerType()))\\\n",
    ".withColumn('probability', F.col('detailed_prediction.probabilities.1'))\\\n",
    ".withColumn('rating', ratings_func((F.col('probability'))))\\\n",
    ".withColumn('predict', predict_func(F.col('predict_int')))\\\n",
    ".drop('detailed_prediction', 'prediction', 'predict_int', 'partition_0')\n",
    "\n",
    "first_cols = ['pclass', 'embarked', 'cabine_prefix', 'ticket_str', 'nametitle']\n",
    "other_cols = sorted([c for c in predict_titanic.columns if c not in first_cols])\n",
    "rearanged_cols = first_cols + other_cols\n",
    "predict_titanic = predict_titanic.select(rearanged_cols)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Alterar o código final que será executado no Glue Jobs. \n",
    "#### ATENÇÃO: Não executar esse código, pois vai dar erro. Ele irá funcionar somente no Glue Jobs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import sys\n",
    "from awsglue.transforms import *\n",
    "from awsglue.utils import getResolvedOptions\n",
    "from awsglue.context import GlueContext\n",
    "from awsglue.job import Job\n",
    "from pyspark.context import SparkContext\n",
    "from pyspark.sql.types import *\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.window import Window\n",
    "from pysparkling.ml import H2OMOJOSettings\n",
    "from pysparkling.ml import H2OMOJOModel\n",
    "import pandas as pd\n",
    "\n",
    "# @params: [JOB_NAME]\n",
    "args = getResolvedOptions(sys.argv, [\"JOB_NAME\"])\n",
    "\n",
    "#Job setup\n",
    "sc = SparkContext()\n",
    "glueContext = GlueContext(sc)\n",
    "spark = glueContext.spark_session\n",
    "spark.conf.set(\"spark.sql.legacy.parquet.int96RebaseModeInRead\", \"CORRECTED\")\n",
    "spark.conf.set(\"spark.sql.legacy.parquet.int96RebaseModeInWrite\", \"CORRECTED\")\n",
    "spark.conf.set(\"spark.sql.legacy.parquet.datetimeRebaseModeInRead\", \"CORRECTED\")\n",
    "spark.conf.set(\"spark.sql.legacy.parquet.datetimeRebaseModeInWrite\", \"CORRECTED\")\n",
    "\n",
    "job = Job(glueContext)\n",
    "job.init(args[\"JOB_NAME\"], args)\n",
    "\n",
    "datasource_titanic = glueContext.create_dynamic_frame.from_catalog(database = \"auladeploymodelos\", table_name = \"titanic_propensity_survive\", transformation_ctx = \"datasource_titanic_propensity_survive\", additional_options={\"mergeSchema\": \"true\"}).toDF()\\\n",
    ".fillna(value=\"Missing\")\n",
    "\n",
    "#Name\n",
    "ModelName = 'titanic_propensity_survive'\n",
    "# ATENÇÃO: nome do bucket criado no S3 (altere para o bucket com o seu nome)\n",
    "bucketName = 'aula-deploy-modelos-seu-nome'\n",
    "\n",
    "#ATENÇÃO\n",
    "#Remover ./output_model/models/best/ da variável PathModelMojo e deixar somente o nome do arquivo .zip\n",
    "\n",
    "\n",
    "\n",
    "############################################ COLAR O CÓDIGO DO DEPLOY SOMENTE ABAIXO DESSA LINHA ###########################\n",
    "\n",
    "#ModelPath\n",
    "PathModelMojo='StackedEnsemble_BestOfFamily_4_AutoML_1_20221011_230015.zip'\n",
    "\n",
    "## Selected Features: variaveis que entraram no modelo\n",
    "CAT = [\n",
    "'pclass'\n",
    ",'embarked'\n",
    ",'cabine_prefix'\n",
    ",'ticket_str'\n",
    ",'nametitle']\n",
    "\n",
    "#float\n",
    "NUM = [\n",
    "'fare'\n",
    ",'sibsp'\n",
    ",'parch'\n",
    ",'age_mean'\n",
    ",'ticket_int']\n",
    "selected_features = CAT + NUM\n",
    "\n",
    "##functions\n",
    "def ratings(p1):\n",
    "    if p1 <= 0.2508362656036639:\n",
    "        return 1\n",
    "    elif p1 <= 0.6540492277407066:\n",
    "        return 2\n",
    "    else:\n",
    "        return 3\n",
    "ratings_func = F.udf(ratings, StringType())\n",
    "\n",
    "def predict_survive(predict):\n",
    "    if predict == 0:\n",
    "        return 'Not survive'\n",
    "    elif predict == 1:\n",
    "        return 'Survive'\n",
    "    else:\n",
    "        return 'predict_ERROR'\n",
    "predict_func = F.udf(predict_survive, StringType())\n",
    "\n",
    "## Escorar a base, criar o score e o rating\n",
    "settings = H2OMOJOSettings(convertUnknownCategoricalLevelsToNa = True, convertInvalidNumbersToNa = True)\n",
    "model = H2OMOJOModel.createFromMojo(PathModelMojo, settings)\n",
    "predict_titanic_1 = model.transform(datasource_titanic)\n",
    "\n",
    "predict_titanic = predict_titanic_1.withColumn('predict_int', F.col('detailed_prediction.label').cast(IntegerType()))\\\n",
    ".withColumn('probability', F.col('detailed_prediction.probabilities.1'))\\\n",
    ".withColumn('rating', ratings_func((F.col('probability'))))\\\n",
    ".withColumn('predict', predict_func(F.col('predict_int')))\\\n",
    ".drop('detailed_prediction', 'prediction', 'predict_int', 'partition_0')\n",
    "\n",
    "first_cols = ['pclass', 'embarked', 'cabine_prefix', 'ticket_str', 'nametitle']\n",
    "other_cols = sorted([c for c in predict_titanic.columns if c not in first_cols])\n",
    "rearanged_cols = first_cols + other_cols\n",
    "predict_titanic = predict_titanic.select(rearanged_cols)\n",
    "\n",
    "############################################# FIM DO BLOCO PARA COLOCAR SEU CÓDIGO #############################################\n",
    "\n",
    "\n",
    "\n",
    "## Salvar no Lakehouse no format parquet\n",
    "predict_titanic.write.mode(\"overwrite\").format(\"parquet\").partitionBy('embarked', 'pclass').save('s3://%s/databases/%s' % (bucketName, ModelName+'_spark'))\n",
    "\n",
    "job.commit()\n",
    "\n",
    "#fim"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.1 Copiar e colar o conteúdo inteiro do bloco a cima e salvar no arqwuivo ./deploy_glue/titanic_predict_glue.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. Fazer upload para o S3 dos arquivos que serão usados pelo Glue jobs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3_resource = my_boto3_session.resource('s3')\n",
    "s3_resource.meta.client.upload_file(PathModelMojo, bucketName, f'deploy/{PathModelMojo.split(\"/\")[-1]}')\n",
    "s3_resource.meta.client.upload_file('./deploy_glue/sparkling-water-assembly-scoring_2.12-3.36.1.1-1-3.1-all.jar', bucketName, 'deploy/sparkling-water-assembly-scoring_2.12-3.36.1.1-1-3.1-all.jar')\n",
    "s3_resource.meta.client.upload_file('./deploy_glue/titanic_predict_glue.py', bucketName, 'deploy/titanic_predict_glue.py')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 6. Criar e executar o Glue Job"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\n",
      "    \"Name\": \"TitanicJob\",\n",
      "    \"ResponseMetadata\": {\n",
      "        \"HTTPHeaders\": {\n",
      "            \"connection\": \"keep-alive\",\n",
      "            \"content-length\": \"21\",\n",
      "            \"content-type\": \"application/x-amz-json-1.1\",\n",
      "            \"date\": \"Fri, 21 Oct 2022 20:21:14 GMT\",\n",
      "            \"x-amzn-requestid\": \"f944ef9e-7390-49dd-9521-21ee981cd852\"\n",
      "        },\n",
      "        \"HTTPStatusCode\": 200,\n",
      "        \"RequestId\": \"f944ef9e-7390-49dd-9521-21ee981cd852\",\n",
      "        \"RetryAttempts\": 0\n",
      "    }\n",
      "}\n"
     ]
    }
   ],
   "source": [
    "glue = my_boto3_session.client('glue', region_name=\"us-east-1\")\n",
    "\n",
    "#Excluir o job antes de criar caso ele já exista\n",
    "response = glue.delete_job(\n",
    "    JobName='TitanicJob'\n",
    ")\n",
    "\n",
    "#Criar o job com o script definido em titanic_predict_glue.py\n",
    "response_create = glue.create_job(\n",
    "    Name='TitanicJob',    \n",
    "    Role= get_execution_role(),\n",
    "    Command={\n",
    "        'Name': 'glueetl',\n",
    "        'ScriptLocation': f's3://{bucketName}/deploy/titanic_predict_glue.py',\n",
    "        'PythonVersion': '3'\n",
    "    },\n",
    "    DefaultArguments={\n",
    "      '--TempDir': f's3://{bucketName}/deploy',\n",
    "      '--job-bookmark-option': 'job-bookmark-disable',\n",
    "      '--additional-python-modules':   'h2o-pysparkling-3.1==3.36.1.3.post1',\n",
    "      '--extra-jars':  f's3://{bucketName}/deploy/sparkling-water-assembly-scoring_2.12-3.36.1.1-1-3.1-all.jar',\n",
    "      '--extra-files':  f's3://{bucketName}/deploy/{PathModelMojo.split(\"/\")[-1]}'\n",
    "    },\n",
    "    MaxRetries=0,\n",
    "    GlueVersion='3.0',\n",
    "    NumberOfWorkers=2,\n",
    "    WorkerType='Standard',\n",
    "      Tags={\n",
    "          'cost_by_tags': 'glue-jobs-titanic'\n",
    "      },\n",
    ")\n",
    "\n",
    "print(json.dumps(response_create, indent=4, sort_keys=True, default=str))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'JobRunId': 'jr_0bb94379c0f54e506b6b1191062930cf5c0de769b22766fca308f71b4f2fca62',\n",
       " 'ResponseMetadata': {'RequestId': '589feb13-a40a-4408-9a2d-06baf3503b72',\n",
       "  'HTTPStatusCode': 200,\n",
       "  'HTTPHeaders': {'date': 'Fri, 21 Oct 2022 20:21:14 GMT',\n",
       "   'content-type': 'application/x-amz-json-1.1',\n",
       "   'content-length': '82',\n",
       "   'connection': 'keep-alive',\n",
       "   'x-amzn-requestid': '589feb13-a40a-4408-9a2d-06baf3503b72'},\n",
       "  'RetryAttempts': 0}}"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "response_run = glue.start_job_run(\n",
    "    JobName=response_create['Name']\n",
    ")\n",
    "response_run"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Job Status: RUNNING\n",
      "Job Status: RUNNING\n",
      "Job Status: RUNNING\n",
      "Job Status: RUNNING\n",
      "Job Status: RUNNING\n",
      "Job Status: RUNNING\n",
      "Job Status: SUCCEEDED\n"
     ]
    }
   ],
   "source": [
    "#Aguardar até o job ser concluido\n",
    "JobRunState='RUNNING'\n",
    "\n",
    "while JobRunState == 'RUNNING':    \n",
    "    response_get = glue.get_job_run(\n",
    "        JobName=response_create['Name'],\n",
    "        RunId=response_run['JobRunId'],\n",
    "        PredecessorsIncluded=True\n",
    "    )\n",
    "    JobRunState = response_get['JobRun']['JobRunState']\n",
    "    print('Job Status: ' + JobRunState)\n",
    "    time.sleep(20)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'ResponseMetadata': {'RequestId': '9668d3d7-4529-43db-a06c-12982c4cccf5',\n",
       "  'HTTPStatusCode': 200,\n",
       "  'HTTPHeaders': {'date': 'Fri, 21 Oct 2022 20:23:38 GMT',\n",
       "   'content-type': 'application/x-amz-json-1.1',\n",
       "   'content-length': '2',\n",
       "   'connection': 'keep-alive',\n",
       "   'x-amzn-requestid': '9668d3d7-4529-43db-a06c-12982c4cccf5'},\n",
       "  'RetryAttempts': 0}}"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Usar o mesmo crawler criado no código 1_Data_Prep.ipynb para adiconar a nova tablea\n",
    "glue.update_crawler(Name='aula-deploy-modelos',\n",
    "                    Targets={'S3Targets': [{'Path': 's3://%s/databases/%s' % (bucketName, ModelName+'_spark'),\n",
    "                               'Exclusions': []}]})\n",
    "\n",
    "glue.start_crawler(Name='aula-deploy-modelos')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 7. Verificar o resultado no Athena\n",
    "#### 7.1. Voltar para a página da AWS e em *Services*, no canto superior esquerdo procure por *Athena*\n",
    "#### 7.2. Em data base, no canto superior esquerdo, selecione auladeploymodelos\n",
    "#### 7.3. Clique sobre a tabela titanic_propensity_survive_spark e selecione a opção \"Preview Table\"\n",
    "#### 7.4. Confira se os dados da tabela estão ok"
   ]
  }
 ],
 "metadata": {
  "instance_type": "ml.t3.medium",
  "kernelspec": {
   "display_name": "Python3 (aula-deploy-modelos/1)",
   "language": "python",
   "name": "Python3__SAGEMAKER_INTERNAL__arn:aws:sagemaker:us-east-1:702113447940:image-version/aula-deploy-modelos/1"
  },
  "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.12"
  },
  "vscode": {
   "interpreter": {
    "hash": "5109d816b82be14675a6b11f8e0f0d2e80f029176ed3710d54e125caa8520dfd"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}