CreateDataSet.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "import sagemaker\n",
    "import pandas as pd\n",
    "import os\n",
    "import numpy as np\n",
    "import json\n",
    "import boto3\n",
    "import string\n",
    "import collections\n",
    "import logging\n",
    "from sklearn.feature_extraction.text import TfidfVectorizer, CountVectorizer\n",
    "from sklearn.cluster import KMeans\n",
    "from sklearn.decomposition import PCA\n",
    "import matplotlib.pyplot as plt\n",
    "from matplotlib import rcParams\n",
    "import matplotlib.cm as cm\n",
    "%matplotlib inline\n",
    "\n",
    "import sagemaker\n",
    "from sagemaker.amazon.amazon_estimator import get_image_uri\n",
    "\n",
    "print(f\"sagemaker:version:{sagemaker.__version__}\")\n",
    "print(f\"Pandas:version:{pd.__version__}\")\n",
    "print(f\"Numpy:version:{np.__version__}\")\n",
    "import logging\n",
    "import seaborn as sns\n",
    "import matplotlib.pyplot as plt\n",
    "import warnings\n",
    "warnings.filterwarnings('ignore')\n",
    "\n",
    "installViaNotebook=False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "import sagemaker\n",
    "\n",
    "\n",
    "region = boto3.Session().region_name\n",
    "role = sagemaker.get_execution_role()\n",
    "#sm_session = sagemaker.session.Session(default_bucket=\"sagemaker-grewaltempl\")\n",
    "#-- try the cross account one \n",
    "sm_session = sagemaker.session.Session(default_bucket=\"sagemaker-crossaccnt-train\")\n",
    "default_bucket = sm_session.default_bucket()\n",
    "sm_client=sm_session.sagemaker_client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Generate Data\n",
    "Generate synthetic housing data -- with no Categorical columns for now -- \n",
    "Generate Json File\n",
    "Generate Zip file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "from random import choice\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "\n",
    "NUM_HOUSES_PER_LOCATION = 1000\n",
    "LOCATIONS = [\n",
    "    \"NewYork_NY\",\n",
    "    \"LosAngeles_CA\",\n",
    "    \"Chicago_IL\",\n",
    "    \"Houston_TX\",\n",
    "    \"Dallas_TX\",\n",
    "    \"Phoenix_AZ\",\n",
    "    \"Philadelphia_PA\",\n",
    "    \"SanAntonio_TX\",\n",
    "    \"SanDiego_CA\",\n",
    "    \"SanFrancisco_CA\",\n",
    "]\n",
    "MAX_YEAR = 2019\n",
    "\n",
    "\n",
    "def generate_price(house):\n",
    "    \"\"\"Generate price based on features of the house\"\"\"\n",
    "\n",
    "    if house[\"FRONT_PORCH\"] == \"y\":\n",
    "        garage = 1\n",
    "    else:\n",
    "        garage = 0\n",
    "\n",
    "    if house[\"FRONT_PORCH\"] == \"y\":\n",
    "        front_porch = 1\n",
    "    else:\n",
    "        front_porch = 0\n",
    "\n",
    "    price = int(\n",
    "        150 * house[\"SQUARE_FEET\"]\n",
    "        + 10000 * house[\"NUM_BEDROOMS\"]\n",
    "        + 15000 * house[\"NUM_BATHROOMS\"]\n",
    "        + 15000 * house[\"LOT_ACRES\"]\n",
    "        + 10000 * garage\n",
    "        + 10000 * front_porch\n",
    "        + 15000 * house[\"GARAGE_SPACES\"]\n",
    "        - 5000 * (MAX_YEAR - house[\"YEAR_BUILT\"])\n",
    "    )\n",
    "    return price\n",
    "\n",
    "\n",
    "def generate_yes_no():\n",
    "    \"\"\"Generate values (y/n) for categorical features\"\"\"\n",
    "    answer = choice([1, 0])\n",
    "    return answer\n",
    "\n",
    "\n",
    "def generate_random_house():\n",
    "    \"\"\"Generate a row of data (single house information)\"\"\"\n",
    "    house = {\n",
    "        \"SHARD_PREFIX\": np.random.randint(0, 4),\n",
    "        \"SQUARE_FEET\": np.random.normal(3000, 750),\n",
    "        \"NUM_BEDROOMS\": np.random.randint(2, 7),\n",
    "        \"NUM_BATHROOMS\": np.random.randint(2, 7) / 2,\n",
    "        \"LOT_ACRES\": round(np.random.normal(1.0, 0.25), 2),\n",
    "        \"GARAGE_SPACES\": np.random.randint(0, 4),\n",
    "        \"YEAR_BUILT\": min(MAX_YEAR, int(np.random.normal(1995, 10))),\n",
    "        \"FRONT_PORCH\": generate_yes_no(),\n",
    "        \"DECK\": generate_yes_no(),\n",
    "    }\n",
    "\n",
    "    price = generate_price(house)\n",
    "\n",
    "    return [\n",
    "        house[\"SHARD_PREFIX\"],\n",
    "        house[\"YEAR_BUILT\"],\n",
    "        house[\"SQUARE_FEET\"],\n",
    "        house[\"NUM_BEDROOMS\"],\n",
    "        house[\"NUM_BATHROOMS\"],\n",
    "        house[\"LOT_ACRES\"],\n",
    "        house[\"GARAGE_SPACES\"],\n",
    "        house[\"FRONT_PORCH\"],\n",
    "        house[\"DECK\"],\n",
    "        price,\n",
    "    ]\n",
    "\n",
    "\n",
    "def generate_houses(num_houses):\n",
    "    \"\"\"Generate housing dataset\"\"\"\n",
    "    house_list = []\n",
    "\n",
    "    for _ in range(num_houses):\n",
    "        house_list.append(generate_random_house())\n",
    "        #print(house_list)\n",
    "\n",
    "    df = pd.DataFrame(\n",
    "        house_list,\n",
    "        columns=[\n",
    "            \"SHARD_PREFIX\",\n",
    "            \"YEAR_BUILT\",\n",
    "            \"SQUARE_FEET\",\n",
    "            \"NUM_BEDROOMS\",\n",
    "            \"NUM_BATHROOMS\",\n",
    "            \"LOT_ACRES\",\n",
    "            \"GARAGE_SPACES\",\n",
    "            \"FRONT_PORCH\",\n",
    "            \"DECK\",\n",
    "            \"PRICE\",\n",
    "        ],\n",
    "    )\n",
    "    return df\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "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>SHARD_PREFIX</th>\n",
       "      <th>YEAR_BUILT</th>\n",
       "      <th>SQUARE_FEET</th>\n",
       "      <th>NUM_BEDROOMS</th>\n",
       "      <th>NUM_BATHROOMS</th>\n",
       "      <th>LOT_ACRES</th>\n",
       "      <th>GARAGE_SPACES</th>\n",
       "      <th>FRONT_PORCH</th>\n",
       "      <th>DECK</th>\n",
       "      <th>PRICE</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>3</td>\n",
       "      <td>1983</td>\n",
       "      <td>3115.372734</td>\n",
       "      <td>3</td>\n",
       "      <td>3.0</td>\n",
       "      <td>0.91</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>375955</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>0</td>\n",
       "      <td>1993</td>\n",
       "      <td>2763.637310</td>\n",
       "      <td>2</td>\n",
       "      <td>1.5</td>\n",
       "      <td>0.84</td>\n",
       "      <td>2</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>369645</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1</td>\n",
       "      <td>1976</td>\n",
       "      <td>3706.016166</td>\n",
       "      <td>5</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.58</td>\n",
       "      <td>2</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>444602</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>3</td>\n",
       "      <td>2010</td>\n",
       "      <td>1550.430927</td>\n",
       "      <td>4</td>\n",
       "      <td>1.5</td>\n",
       "      <td>1.04</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>265664</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>1</td>\n",
       "      <td>1990</td>\n",
       "      <td>3528.656218</td>\n",
       "      <td>2</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.12</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>451098</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>2</td>\n",
       "      <td>1976</td>\n",
       "      <td>2942.815192</td>\n",
       "      <td>3</td>\n",
       "      <td>2.5</td>\n",
       "      <td>0.88</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>322122</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>0</td>\n",
       "      <td>2008</td>\n",
       "      <td>2561.264003</td>\n",
       "      <td>4</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.81</td>\n",
       "      <td>3</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>441339</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>3</td>\n",
       "      <td>2002</td>\n",
       "      <td>2277.990075</td>\n",
       "      <td>4</td>\n",
       "      <td>2.5</td>\n",
       "      <td>1.20</td>\n",
       "      <td>2</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>382198</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>3</td>\n",
       "      <td>2003</td>\n",
       "      <td>3141.150092</td>\n",
       "      <td>6</td>\n",
       "      <td>1.5</td>\n",
       "      <td>1.07</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>504722</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>0</td>\n",
       "      <td>2019</td>\n",
       "      <td>4050.054475</td>\n",
       "      <td>6</td>\n",
       "      <td>2.0</td>\n",
       "      <td>0.68</td>\n",
       "      <td>2</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>737708</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   SHARD_PREFIX  YEAR_BUILT  SQUARE_FEET  NUM_BEDROOMS  NUM_BATHROOMS  \\\n",
       "0             3        1983  3115.372734             3            3.0   \n",
       "1             0        1993  2763.637310             2            1.5   \n",
       "2             1        1976  3706.016166             5            1.0   \n",
       "3             3        2010  1550.430927             4            1.5   \n",
       "4             1        1990  3528.656218             2            1.0   \n",
       "5             2        1976  2942.815192             3            2.5   \n",
       "6             0        2008  2561.264003             4            1.0   \n",
       "7             3        2002  2277.990075             4            2.5   \n",
       "8             3        2003  3141.150092             6            1.5   \n",
       "9             0        2019  4050.054475             6            2.0   \n",
       "\n",
       "   LOT_ACRES  GARAGE_SPACES  FRONT_PORCH  DECK   PRICE  \n",
       "0       0.91              0            0     1  375955  \n",
       "1       0.84              2            1     1  369645  \n",
       "2       0.58              2            0     1  444602  \n",
       "3       1.04              0            0     0  265664  \n",
       "4       1.12              1            1     1  451098  \n",
       "5       0.88              1            1     0  322122  \n",
       "6       0.81              3            0     1  441339  \n",
       "7       1.20              2            0     0  382198  \n",
       "8       1.07              1            1     0  504722  \n",
       "9       0.68              2            1     0  737708  "
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = generate_houses(10)\n",
    "df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[3, 0, 1, 3, 1, 2, 0, 3, 3, 0]\n"
     ]
    }
   ],
   "source": [
    "print(df['SHARD_PREFIX'].to_list())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 129,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "S3:::s3_data=s3://sagemaker-crossaccnt-train/data/finance/distrib::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard/sh0/test_0.csv\n",
      "S3:::s3_data=s3://sagemaker-crossaccnt-train/data/finance/distrib::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard/sh1/test_1.csv\n",
      "S3:::s3_data=s3://sagemaker-crossaccnt-train/data/finance/distrib::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard/sh2/test_2.csv\n",
      "S3:::s3_data=s3://sagemaker-crossaccnt-train/data/finance/distrib::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard/sh3/test_3.csv\n"
     ]
    }
   ],
   "source": [
    "for shard in range(0,4):\n",
    "    df_shard = df[df['SHARD_PREFIX'] == shard]\n",
    "    if len(df_shard) > 0:\n",
    "        data_loc = f\"./data/test_{shard}.csv\"\n",
    "        s3_uri = f\"s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard/sh{shard}/\"\n",
    "        df_shard.to_csv(data_loc, na_rep=0 )\n",
    "\n",
    "        s3_upload_uri = sagemaker.s3.S3Uploader.upload(\n",
    "            local_path=data_loc,\n",
    "            desired_s3_uri=s3_uri,\n",
    "            kms_key=None,\n",
    "            sagemaker_session=sm_session\n",
    "        )\n",
    "        print(f\"S3:::s3_data={s3_data}::Uploaded:={s3_upload_uri}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Tar ball\n",
    "Create tar ball to be uploaded for testing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "S3:::tar_loc=./data/test_0.tar.gz::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi-tar/shard/test_0.tar.gz\n",
      "S3:::tar_loc=./data/test_1.tar.gz::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi-tar/shard/test_1.tar.gz\n",
      "S3:::tar_loc=./data/test_2.tar.gz::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi-tar/shard/test_2.tar.gz\n",
      "S3:::tar_loc=./data/test_3.tar.gz::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi-tar/shard/test_3.tar.gz\n"
     ]
    }
   ],
   "source": [
    "import tarfile\n",
    "for shard in range(0,4):\n",
    "    df_shard = df[df['SHARD_PREFIX'] == shard]\n",
    "    if len(df_shard) > 0:\n",
    "        data_loc = f\"./data/test_{shard}.csv\"\n",
    "        tar_data_loc = f\"./data/test_{shard}.tar.gz\"\n",
    "        s3_data = f\"s3://sagemaker-crossaccnt-train/data/finance/distrib\"\n",
    "        df_shard.to_csv(data_loc, na_rep=0 )\n",
    "    \n",
    "        tar = tarfile.open(tar_data_loc, \"w:gz\")\n",
    "        for name in [data_loc,]:\n",
    "            tar.add(name)\n",
    "        tar.close()\n",
    "        s3_upload_uri = sagemaker.s3.S3Uploader.upload(\n",
    "            local_path=tar_data_loc,\n",
    "            desired_s3_uri=f\"s3://sagemaker-crossaccnt-train/data/finance/distrib-multi-tar/shard\",\n",
    "            kms_key=None,\n",
    "            sagemaker_session=sm_session\n",
    "        )\n",
    "        print(f\"S3:::tar_loc={tar_data_loc}::Uploaded:={s3_upload_uri}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting ./scripts/distprocess.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile ./scripts/distprocess.py\n",
    "\n",
    "\"\"\"Feature engineers the distributed data set \"\"\"\n",
    "import argparse\n",
    "import logging\n",
    "import os\n",
    "import pathlib\n",
    "import requests\n",
    "import tempfile\n",
    "\n",
    "import boto3\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "import json\n",
    "\n",
    "import multiprocessing\n",
    "\n",
    "_logger = logging.getLogger()\n",
    "_logger.setLevel(logging.INFO)\n",
    "_logger.addHandler(logging.StreamHandler())\n",
    "\n",
    "try:\n",
    "    _logger.info(f\"Pandas:version:{pd.__version__}\")\n",
    "    _logger.info(f\"Numpy:version:{np.__version__}\")\n",
    "except:\n",
    "    pass\n",
    "\n",
    "\n",
    "# Since we get a headerless CSV file we specify the column names here.\n",
    "\n",
    "# Since we get a headerless CSV file we specify the column names here.\n",
    "\n",
    "columns=[\n",
    "    \"SHARD_PREFIX\",\n",
    "    \"YEAR_BUILT\",\n",
    "    \"SQUARE_FEET\",\n",
    "    \"NUM_BEDROOMS\",\n",
    "    \"NUM_BATHROOMS\",\n",
    "    \"LOT_ACRES\",\n",
    "    \"GARAGE_SPACES\",\n",
    "    \"FRONT_PORCH\",\n",
    "    \"DECK\",\n",
    "    \"PRICE\"\n",
    "]\n",
    "\n",
    "\n",
    "columns_dtype = {\n",
    "    'SHARD_PREFIX': np.int, \n",
    "    \"YEAR_BUILT\": np.int,\n",
    "    \"SQUARE_FEET\": np.float64,\n",
    "    \"NUM_BEDROOMS\": np.int,\n",
    "    \"NUM_BATHROOMS\": np.int,\n",
    "    \"LOT_ACRES\": np.float64,\n",
    "    \"GARAGE_SPACES\": np.int,\n",
    "    \"FRONT_PORCH\": np.int,\n",
    "    \"DECK\": np.int,\n",
    "    \"PRICE\": np.float64\n",
    "    #   'body': str, \n",
    "}\n",
    "\n",
    "def merge_two_dicts(x, y):\n",
    "    \"\"\"Merges two dicts, returning a new copy.\"\"\"\n",
    "    z = x.copy()\n",
    "    z.update(y)\n",
    "    return z\n",
    "\n",
    "def listLocalDirectory(dirPath=\".\"):\n",
    "    for path, dnames, fnames in os.walk(dirPath):\n",
    "        _logger.info(f\"List::path={path}::dirNames={dnames}::fileNames={fnames}::\")\n",
    "\n",
    "    print(\"list:done:\")\n",
    "       \n",
    "if __name__ == \"__main__\":\n",
    "    _logger.info(\"Starting distprocess.py.\")\n",
    "    parser = argparse.ArgumentParser()\n",
    "    parser.add_argument(\"--data-size\", type=int, default=100)\n",
    "    parser.add_argument(\"--input-path\", type=str, default='/opt/ml/processing/input') # \"input_path\", \"/opt/ml/processing/input\",\n",
    "    args = parser.parse_args()\n",
    "    data_size = args.data_size\n",
    "    input_dir = args.input_path\n",
    "\n",
    "    # Split list of files into sub-lists\n",
    "    cpu_count = multiprocessing.cpu_count()\n",
    "\n",
    "    \n",
    "    BASE_DIR = \"/opt/ml/processing\"\n",
    "    \n",
    "    print(input_dir)\n",
    "    _logger.info(f\"Input:data:path:={input_dir}::\")\n",
    "\n",
    "\n",
    "    #fn = os.path.join(\"/opt/ml/processing/input\", \"combined_tweets.csv\")\n",
    "    \n",
    "    onlyFiles = [f for f in os.listdir(input_dir) if os.path.isfile(os.path.join(input_dir, f))]\n",
    "    _logger.info(f\"Data Downloaded::Now Reading downloaded data.:dir:{input_dir}::And:FILES:ARE::{onlyFiles}\")\n",
    "    \n",
    "    fn = os.path.join(input_dir, onlyFiles[0])\n",
    "    \n",
    "    # read in csv\n",
    "    df = pd.read_csv(fn, names=columns)\n",
    "    #_logger.info(f\"df:head={df.head()}\")\n",
    "    _logger.info(f\"DF:SHARD:INDEX={df['SHARD_PREFIX'].to_list()}::\")\n",
    "    \n",
    "    _logger.info(f\"AFTER:READ:Going to write it to {BASE_DIR}/output : file will be {BASE_DIR}/output/{onlyFiles[0]}:: \")\n",
    "    _logger.info(f\"READ:Data:len={len(df)}::  shape={df.shape}::\")\n",
    "    df.to_csv(\n",
    "        f\"{BASE_DIR}/output/{onlyFiles[0]}\", header=True, index=False\n",
    "    )\n",
    "    \n",
    "    _logger.info(\"All Done !! written out !!\")\n",
    "    \n",
    "    # ----------------   OLD TEMPLATE CODE --------------------#"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Write Tar\n",
    "tar processessing script"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting ./scripts/distprocess_tar.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile ./scripts/distprocess_tar.py\n",
    "\n",
    "\"\"\"Feature engineers the distributed data set using tar ball \"\"\"\n",
    "import argparse\n",
    "import logging\n",
    "import os\n",
    "import pathlib\n",
    "import requests\n",
    "import tempfile\n",
    "\n",
    "import boto3\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "import json\n",
    "\n",
    "import multiprocessing\n",
    "import tarfile\n",
    "\n",
    "import pathlib\n",
    "import shutil\n",
    "import traceback\n",
    "\n",
    "\n",
    "_logger = logging.getLogger()\n",
    "_logger.setLevel(logging.INFO)\n",
    "_logger.addHandler(logging.StreamHandler())\n",
    "\n",
    "try:\n",
    "    _logger.info(f\"Pandas:version:{pd.__version__}\")\n",
    "    _logger.info(f\"Numpy:version:{np.__version__}\")\n",
    "except:\n",
    "    pass\n",
    "\n",
    "\n",
    "# Since we get a headerless CSV file we specify the column names here.\n",
    "\n",
    "# Since we get a headerless CSV file we specify the column names here.\n",
    "\n",
    "columns=[\n",
    "    \"SHARD_PREFIX\",\n",
    "    \"YEAR_BUILT\",\n",
    "    \"SQUARE_FEET\",\n",
    "    \"NUM_BEDROOMS\",\n",
    "    \"NUM_BATHROOMS\",\n",
    "    \"LOT_ACRES\",\n",
    "    \"GARAGE_SPACES\",\n",
    "    \"FRONT_PORCH\",\n",
    "    \"DECK\",\n",
    "    \"PRICE\"\n",
    "]\n",
    "\n",
    "\n",
    "columns_dtype = {\n",
    "    'SHARD_PREFIX': np.int, \n",
    "    \"YEAR_BUILT\": np.int,\n",
    "    \"SQUARE_FEET\": np.float64,\n",
    "    \"NUM_BEDROOMS\": np.int,\n",
    "    \"NUM_BATHROOMS\": np.int,\n",
    "    \"LOT_ACRES\": np.float64,\n",
    "    \"GARAGE_SPACES\": np.int,\n",
    "    \"FRONT_PORCH\": np.int,\n",
    "    \"DECK\": np.int,\n",
    "    \"PRICE\": np.float64\n",
    "    #   'body': str, \n",
    "}\n",
    "\n",
    "def merge_two_dicts(x, y):\n",
    "    \"\"\"Merges two dicts, returning a new copy.\"\"\"\n",
    "    z = x.copy()\n",
    "    z.update(y)\n",
    "    return z\n",
    "\n",
    "def listLocalDirectory(dirPath=\".\"):\n",
    "    for path, dnames, fnames in os.walk(dirPath):\n",
    "        _logger.info(f\"List::path={path}::dirNames={dnames}::fileNames={fnames}::\")\n",
    "\n",
    "    print(\"list:done:\")\n",
    "       \n",
    "if __name__ == \"__main__\":\n",
    "    _logger.info(\"Starting distprocess.py.\")\n",
    "    parser = argparse.ArgumentParser()\n",
    "    parser.add_argument(\"--data-size\", type=int, default=100)\n",
    "    parser.add_argument(\"--input-path\", type=str, default='/opt/ml/processing/input') # \"input_path\", \"/opt/ml/processing/input\",\n",
    "    args = parser.parse_args()\n",
    "    data_size = args.data_size\n",
    "    input_dir = args.input_path\n",
    "    \n",
    "\n",
    "    for arg, value in sorted(vars(args).items()):\n",
    "        _logger.info(f\"TAR:logging:all:environment:variables={arg}::value={value}::\")\n",
    "\n",
    "    print(\"printing the env vars now\")\n",
    "    env_var = os.environ    \n",
    "    import pprint\n",
    "    pprint.pprint(dict(env_var), width = 1)\n",
    "    \n",
    "    # Split list of files into sub-lists\n",
    "    cpu_count = multiprocessing.cpu_count()\n",
    "    \n",
    "    BASE_DIR = \"/opt/ml/processing\"\n",
    "    \n",
    "    print(input_dir)\n",
    "    _logger.info(f\"Input:data:path:={input_dir}::\")\n",
    "\n",
    "    try:\n",
    "        with open('/opt/ml/input/config/inputdataconfig.json') as f:\n",
    "            lines = f.readlines()\n",
    "            print(lines)\n",
    "    except:\n",
    "        print(\"error in reading input data config\")\n",
    "        \n",
    "    #fn = os.path.join(\"/opt/ml/processing/input\", \"combined_tweets.csv\")\n",
    "    \n",
    "    onlyFiles = [f for f in os.listdir(input_dir) if os.path.isfile(os.path.join(input_dir, f))]\n",
    "    _logger.info(f\"TAR:Data Downloaded::Now Reading downloaded data.:dir:{input_dir}::And:FILES:ARE::{onlyFiles}\")\n",
    "    \n",
    "    # -- ITERATE OVER EACH OF THE TAR FILES which we got On this Instance and process them \n",
    "    for tarFiles in onlyFiles:\n",
    "        # -- Read the TAR file, extract to temp proc directory , then pick up the first file to process it \n",
    "        fn = os.path.join(input_dir, tarFiles) #onlyFiles[0])\n",
    "        proc_dir = f\"{input_dir}/filesProc\"\n",
    "        pathlib.Path(proc_dir).mkdir(parents=True, exist_ok=True)\n",
    "\n",
    "        _logger.info(f\"TAR:Creating directory:actual Procesing dir={proc_dir}:file_loc={fn}:\")\n",
    "\n",
    "        # -- extract the TAR to the ./filesProc/data which is the temp location.\n",
    "        with tarfile.open(fn, 'r:*') as s3_tar:\n",
    "            s3_tar.extractall(proc_dir)\n",
    "\n",
    "        _logger.info(f\"TAR FILE Extracted successfully to dir={proc_dir}\")  \n",
    "        allFiles = [f for f in os.listdir(proc_dir) ]\n",
    "        _logger.info(f\"TAR FILE listing :{proc_dir}: gives us :{allFiles}:\") \n",
    "\n",
    "        # -- Create the location for pandas to read the EXTRACTED CSV file  - similiar to ./filesProc/data/test_01.csv\n",
    "        proc_data_dir = os.path.join(proc_dir, \"data\")\n",
    "        _logger.info(f\"TAR FILE :proc_data_dir:to:{proc_data_dir}::\")\n",
    "        onlyFiles = [f for f in os.listdir(proc_data_dir) if os.path.isfile(os.path.join(proc_data_dir, f))]\n",
    "        _logger.info(f\"TAR: After extraction:READ::dir:{proc_data_dir}::And:FILES:ARE::{onlyFiles}\")\n",
    "        fn = os.path.join(proc_data_dir, onlyFiles[0])\n",
    "        _logger.info(f\"TAR: After extraction:FILE:loc={fn}::\")\n",
    "\n",
    "        # -- Read in csv at ./filesProc/data/test_01.csv\n",
    "        df = pd.read_csv(fn, names=columns)\n",
    "        _logger.info(f\"DF:SHARD:INDEX={df['SHARD_PREFIX'].to_list()}::\")\n",
    "\n",
    "        # -- DO some processing on this CSV file read \n",
    "        _logger.info(f\"AFTER:READ:Going to write it to {BASE_DIR}/output : file will be {BASE_DIR}/output/{onlyFiles[0]}:: \")\n",
    "        _logger.info(f\"READ:Data:len={len(df)}::  shape={df.shape}::\")\n",
    "        df.to_csv(\n",
    "            f\"{BASE_DIR}/output/{onlyFiles[0]}\", header=True, index=False\n",
    "        )\n",
    "\n",
    "        # -- Clean up the temp diectories ./filesProc so we can repeat for the other files\n",
    "        try:\n",
    "            shutil.rmtree(proc_dir)\n",
    "            _logger.info(f\"Temp directory={proc_dir}:: deleted\")\n",
    "        except :\n",
    "            _logger.info(f\"Error: in deleting the TEMP dir={proc_dir}::using traceback={traceback.format_exc()}::\")\n",
    "\n",
    "        \n",
    "    _logger.info(\"All Done !! written out !!\")\n",
    "    \n",
    "    # ----------------   OLD TEMPLATE CODE --------------------#"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Role:SageMaker:session=arn:aws:iam::622343165275:role/service-role/AmazonSageMaker-ExecutionRole-20220208T115633::\n"
     ]
    }
   ],
   "source": [
    "region = boto3.Session().region_name\n",
    "role = sagemaker.get_execution_role()\n",
    "print(f\"Role:SageMaker:session={role}::\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Run \n",
    "Run the pre processingjob with shardbys3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Using:role=arn:aws:iam::622343165275:role/service-role/AmazonSageMaker-ExecutionRole-20220208T115633:\n",
      "using SageMaker session=<sagemaker.session.Session object at 0x7f5ef5bbf790>:\n",
      "Preprocessing:script:loc=./scripts/distprocess.py:\n",
      "\n",
      "Job Name:  dist-proc-2022-06-02-03-22-49-879\n",
      "Inputs:  [{'InputName': 'input-1', 'AppManaged': False, 'S3Input': {'S3Uri': 's3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard', 'LocalPath': '/opt/ml/processing/input', 'S3DataType': 'S3Prefix', 'S3InputMode': 'File', 'S3DataDistributionType': 'ShardedByS3Key', 'S3CompressionType': 'None'}}, {'InputName': 'code', 'AppManaged': False, 'S3Input': {'S3Uri': 's3://sagemaker-crossaccnt-train/dist-proc-2022-06-02-03-22-49-879/input/code/distprocess.py', 'LocalPath': '/opt/ml/processing/input/code', 'S3DataType': 'S3Prefix', 'S3InputMode': 'File', 'S3DataDistributionType': 'FullyReplicated', 'S3CompressionType': 'None'}}]\n",
      "Outputs:  [{'OutputName': 'final_output', 'AppManaged': False, 'S3Output': {'S3Uri': 's3://sagemaker-crossaccnt-train/data/finance/distrib-multi/output', 'LocalPath': '/opt/ml/processing/output', 'S3UploadMode': 'EndOfJob'}}]\n",
      "..........................\u001b[35mPandas:version:1.1.3\u001b[0m\n",
      "\u001b[35mNumpy:version:1.21.0\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess.py:48: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  'SHARD_PREFIX': np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess.py:49: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"YEAR_BUILT\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess.py:51: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"NUM_BEDROOMS\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess.py:52: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"NUM_BATHROOMS\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess.py:54: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"GARAGE_SPACES\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess.py:55: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"FRONT_PORCH\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess.py:56: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"DECK\": np.int,\u001b[0m\n",
      "\u001b[35mStarting distprocess.py.\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input\u001b[0m\n",
      "\u001b[35mInput:data:path:=/opt/ml/processing/input::\u001b[0m\n",
      "\u001b[35mData Downloaded::Now Reading downloaded data.:dir:/opt/ml/processing/input::And:FILES:ARE::[]\u001b[0m\n",
      "\u001b[35mTraceback (most recent call last):\n",
      "  File \"/opt/ml/processing/input/code/distprocess.py\", line 97, in <module>\n",
      "    fn = os.path.join(input_dir, onlyFiles[0])\u001b[0m\n",
      "\u001b[35mIndexError: list index out of range\u001b[0m\n",
      "\u001b[34mPandas:version:1.1.3\u001b[0m\n",
      "\u001b[34mNumpy:version:1.21.0\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess.py:48: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  'SHARD_PREFIX': np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess.py:49: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"YEAR_BUILT\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess.py:51: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"NUM_BEDROOMS\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess.py:52: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"NUM_BATHROOMS\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess.py:54: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"GARAGE_SPACES\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess.py:55: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"FRONT_PORCH\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess.py:56: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"DECK\": np.int,\u001b[0m\n",
      "\u001b[34mStarting distprocess.py.\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input\u001b[0m\n",
      "\u001b[34mInput:data:path:=/opt/ml/processing/input::\u001b[0m\n",
      "\u001b[34mData Downloaded::Now Reading downloaded data.:dir:/opt/ml/processing/input::And:FILES:ARE::[]\u001b[0m\n",
      "\u001b[34mTraceback (most recent call last):\n",
      "  File \"/opt/ml/processing/input/code/distprocess.py\", line 97, in <module>\n",
      "    fn = os.path.join(input_dir, onlyFiles[0])\u001b[0m\n",
      "\u001b[34mIndexError: list index out of range\u001b[0m\n",
      "\n"
     ]
    },
    {
     "ename": "UnexpectedStatusException",
     "evalue": "Error for Processing job dist-proc-2022-06-02-03-22-49-879: Failed. Reason: AlgorithmError: See job logs for more information",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mUnexpectedStatusException\u001b[0m                 Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-56-75c7e5e8d6c5>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m     59\u001b[0m     \u001b[0marguments\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mjob_arguments\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     60\u001b[0m     \u001b[0mwait\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mTrue\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 61\u001b[0;31m     \u001b[0mlogs\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m\"All\"\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     62\u001b[0m )\n",
      "\u001b[0;32m/opt/conda/lib/python3.7/site-packages/sagemaker/processing.py\u001b[0m in \u001b[0;36mrun\u001b[0;34m(self, code, inputs, outputs, arguments, wait, logs, job_name, experiment_config, kms_key)\u001b[0m\n\u001b[1;32m    551\u001b[0m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjobs\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mappend\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mlatest_job\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    552\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mwait\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 553\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mlatest_job\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mwait\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mlogs\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mlogs\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    554\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    555\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m_include_code_in_inputs\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0minputs\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcode\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mkms_key\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mNone\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/opt/conda/lib/python3.7/site-packages/sagemaker/processing.py\u001b[0m in \u001b[0;36mwait\u001b[0;34m(self, logs)\u001b[0m\n\u001b[1;32m    960\u001b[0m         \"\"\"\n\u001b[1;32m    961\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mlogs\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 962\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msagemaker_session\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mlogs_for_processing_job\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjob_name\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mwait\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mTrue\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    963\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    964\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msagemaker_session\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mwait_for_processing_job\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjob_name\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/opt/conda/lib/python3.7/site-packages/sagemaker/session.py\u001b[0m in \u001b[0;36mlogs_for_processing_job\u001b[0;34m(self, job_name, wait, poll)\u001b[0m\n\u001b[1;32m   3879\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   3880\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mwait\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 3881\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_check_job_status\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mjob_name\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mdescription\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m\"ProcessingJobStatus\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   3882\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0mdot\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   3883\u001b[0m                 \u001b[0mprint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/opt/conda/lib/python3.7/site-packages/sagemaker/session.py\u001b[0m in \u001b[0;36m_check_job_status\u001b[0;34m(self, job, desc, status_key_name)\u001b[0m\n\u001b[1;32m   3331\u001b[0m                 \u001b[0mmessage\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mmessage\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   3332\u001b[0m                 \u001b[0mallowed_statuses\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m\"Completed\"\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m\"Stopped\"\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 3333\u001b[0;31m                 \u001b[0mactual_status\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mstatus\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   3334\u001b[0m             )\n\u001b[1;32m   3335\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mUnexpectedStatusException\u001b[0m: Error for Processing job dist-proc-2022-06-02-03-22-49-879: Failed. Reason: AlgorithmError: See job logs for more information"
     ]
    }
   ],
   "source": [
    "from sagemaker.sklearn.processing import SKLearnProcessor\n",
    "from sagemaker.workflow.parameters import (\n",
    "    ParameterInteger,\n",
    "    ParameterString,\n",
    ")\n",
    "\n",
    "\n",
    "print(f\"Using:role={role}:\")\n",
    "print(f\"using SageMaker session={sm_session}:\")\n",
    "\n",
    "BASE_JOB_PREFIX=\"smjobs-dist\",  # Choose any name\n",
    "\n",
    "processing_instance_type = ParameterString(\n",
    "        name=\"ProcessingInstanceType\", default_value=\"ml.m5.xlarge\"\n",
    ")\n",
    "\n",
    "\n",
    "# Processing step for feature engineering\n",
    "sklearn_processor = SKLearnProcessor(\n",
    "        framework_version=\"0.23-1\",\n",
    "        instance_type=processing_instance_type,\n",
    "        instance_count=2,\n",
    "        base_job_name=\"dist-proc\" ,\n",
    "        sagemaker_session=sm_session,\n",
    "        role=role,\n",
    "    )\n",
    "\n",
    "from sagemaker.processing import ProcessingInput, ProcessingOutput\n",
    "inputs_p=[\n",
    "    ProcessingInput(\n",
    "        #source=\"s3://sagemaker-crossaccnt-train/data/finance/distrib\",\n",
    "        source=\"s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard\",\n",
    "        destination='/opt/ml/processing/input',\n",
    "        s3_data_distribution_type='ShardedByS3Key'\n",
    "    ),\n",
    " ]\n",
    "outputs_p=[\n",
    "    ProcessingOutput(\n",
    "        s3_upload_mode=\"EndOfJob\",\n",
    "        output_name='final_output',\n",
    "        source='/opt/ml/processing/output',\n",
    "        #destination='s3://sagemaker-crossaccnt-train/data/finance/distrib/output'\n",
    "        destination='s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/output'\n",
    "    ),\n",
    "    \n",
    "    \n",
    "]\n",
    "# -- if we d onot create a output then this directory is never creatd on tbe processing job\n",
    "\n",
    "preproc_script =  \"./scripts/distprocess.py\"\n",
    "\n",
    "print(f\"Preprocessing:script:loc={preproc_script}:\")\n",
    "job_arguments=[\"--input-path\", \"/opt/ml/processing/input\", \n",
    "              \"--data-size\", \"100\"]\n",
    "sklearn_processor.run(\n",
    "    code=preproc_script,\n",
    "    inputs=inputs_p,\n",
    "    outputs=outputs_p,\n",
    "    arguments=job_arguments, \n",
    "    wait=True,\n",
    "    logs=\"All\",\n",
    ")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Run Tar\n",
    "Run the Tar processing job"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Using:role=arn:aws:iam::622343165275:role/service-role/AmazonSageMaker-ExecutionRole-20220208T115633:\n",
      "using SageMaker session=<sagemaker.session.Session object at 0x7f5ef5bbf790>:\n",
      "Preprocessing:script:loc=./scripts/distprocess_tar.py:\n",
      "\n",
      "Job Name:  dist-proc-2022-06-08-16-39-53-410\n",
      "Inputs:  [{'InputName': 'input-1', 'AppManaged': False, 'S3Input': {'S3Uri': 's3://sagemaker-crossaccnt-train/data/finance/distrib-multi-tar/shard', 'LocalPath': '/opt/ml/processing/input', 'S3DataType': 'S3Prefix', 'S3InputMode': 'File', 'S3DataDistributionType': 'ShardedByS3Key', 'S3CompressionType': 'None'}}, {'InputName': 'code', 'AppManaged': False, 'S3Input': {'S3Uri': 's3://sagemaker-crossaccnt-train/dist-proc-2022-06-08-16-39-53-410/input/code/distprocess_tar.py', 'LocalPath': '/opt/ml/processing/input/code', 'S3DataType': 'S3Prefix', 'S3InputMode': 'File', 'S3DataDistributionType': 'FullyReplicated', 'S3CompressionType': 'None'}}]\n",
      "Outputs:  [{'OutputName': 'final_output', 'AppManaged': False, 'S3Output': {'S3Uri': 's3://sagemaker-crossaccnt-train/data/finance/distrib-multi-tar/output', 'LocalPath': '/opt/ml/processing/output', 'S3UploadMode': 'EndOfJob'}}]\n",
      ".............................\u001b[34mPandas:version:1.1.3\u001b[0m\n",
      "\u001b[34mNumpy:version:1.21.0\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess_tar.py:54: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  'SHARD_PREFIX': np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess_tar.py:55: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"YEAR_BUILT\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess_tar.py:57: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"NUM_BEDROOMS\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess_tar.py:58: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"NUM_BATHROOMS\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess_tar.py:60: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"GARAGE_SPACES\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess_tar.py:61: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"FRONT_PORCH\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distprocess_tar.py:62: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"DECK\": np.int,\u001b[0m\n",
      "\u001b[34mStarting distprocess.py.\u001b[0m\n",
      "\u001b[34mTAR:logging:all:environment:variables=data_size::value=100::\u001b[0m\n",
      "\u001b[34mTAR:logging:all:environment:variables=input_path::value=/opt/ml/processing/input::\u001b[0m\n",
      "\u001b[34mprinting the env vars now\u001b[0m\n",
      "\u001b[34m{'AWS_CONTAINER_CREDENTIALS_RELATIVE_URI': '/v2/credentials/RAbFHIwajO4F6oxZpQjR2NNzXdAZu_M2FI5Ig_UGvhc',\n",
      " 'AWS_REGION': 'us-east-1',\n",
      " 'HOME': '/root',\n",
      " 'HOSTNAME': 'ip-10-0-186-56.ec2.internal',\n",
      " 'LANG': 'C.UTF-8',\n",
      " 'LC_ALL': 'C.UTF-8',\n",
      " 'PATH': '/miniconda3/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin',\n",
      " 'PYTHONDONTWRITEBYTECODE': '1',\n",
      " 'PYTHONIOENCODING': 'UTF-8',\n",
      " 'PYTHONUNBUFFERED': '1',\n",
      " 'SAGEMAKER_SERVING_MODULE': 'sagemaker_sklearn_container.serving:main',\n",
      " 'SAGEMAKER_SKLEARN_VERSION': '0.23-1',\n",
      " 'SAGEMAKER_TRAINING_MODULE': 'sagemaker_sklearn_container.training:main',\n",
      " 'SKLEARN_MMS_CONFIG': '/home/model-server/config.properties',\n",
      " 'SM_CHECKPOINT_CONFIG_FILE': '/opt/ml/input/config/checkpointconfig.json',\n",
      " 'SM_INPUT': '/opt/ml/input',\n",
      " 'SM_INPUT_DATA_CONFIG_FILE': '/opt/ml/input/config/inputdataconfig.json',\n",
      " 'SM_INPUT_TRAINING_CONFIG_FILE': '/opt/ml/input/config/hyperparameters.json',\n",
      " 'SM_MODEL_DIR': '/opt/ml/model',\n",
      " 'TEMP': '/home/model-server/tmp'}\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input\u001b[0m\n",
      "\u001b[34mInput:data:path:=/opt/ml/processing/input::\u001b[0m\n",
      "\u001b[34mTAR:Data Downloaded::Now Reading downloaded data.:dir:/opt/ml/processing/input::And:FILES:ARE::['test_2.tar.gz', 'test_0.tar.gz']\u001b[0m\n",
      "\u001b[34mTAR:Creating directory:actual Procesing dir=/opt/ml/processing/input/filesProc:file_loc=/opt/ml/processing/input/test_2.tar.gz:\u001b[0m\n",
      "\u001b[34mTAR FILE Extracted successfully to dir=/opt/ml/processing/input/filesProc\u001b[0m\n",
      "\u001b[34mTAR FILE listing :/opt/ml/processing/input/filesProc: gives us :['data']:\u001b[0m\n",
      "\u001b[34mTAR FILE :proc_data_dir:to:/opt/ml/processing/input/filesProc/data::\u001b[0m\n",
      "\u001b[34mTAR: After extraction:READ::dir:/opt/ml/processing/input/filesProc/data::And:FILES:ARE::['test_2.csv']\u001b[0m\n",
      "\u001b[34mTAR: After extraction:FILE:loc=/opt/ml/processing/input/filesProc/data/test_2.csv::\u001b[0m\n",
      "\u001b[34mDF:SHARD:INDEX=['SHARD_PREFIX', '2']::\u001b[0m\n",
      "\u001b[34mAFTER:READ:Going to write it to /opt/ml/processing/output : file will be /opt/ml/processing/output/test_2.csv:: \u001b[0m\n",
      "\u001b[34mREAD:Data:len=2::  shape=(2, 10)::\u001b[0m\n",
      "\u001b[34mTemp directory=/opt/ml/processing/input/filesProc:: deleted\u001b[0m\n",
      "\u001b[34mTAR:Creating directory:actual Procesing dir=/opt/ml/processing/input/filesProc:file_loc=/opt/ml/processing/input/test_0.tar.gz:\u001b[0m\n",
      "\u001b[34mTAR FILE Extracted successfully to dir=/opt/ml/processing/input/filesProc\u001b[0m\n",
      "\u001b[34mTAR FILE listing :/opt/ml/processing/input/filesProc: gives us :['data']:\u001b[0m\n",
      "\u001b[34mTAR FILE :proc_data_dir:to:/opt/ml/processing/input/filesProc/data::\u001b[0m\n",
      "\u001b[34mTAR: After extraction:READ::dir:/opt/ml/processing/input/filesProc/data::And:FILES:ARE::['test_0.csv']\u001b[0m\n",
      "\u001b[34mTAR: After extraction:FILE:loc=/opt/ml/processing/input/filesProc/data/test_0.csv::\u001b[0m\n",
      "\u001b[34mDF:SHARD:INDEX=['SHARD_PREFIX', '0', '0', '0']::\u001b[0m\n",
      "\u001b[34mAFTER:READ:Going to write it to /opt/ml/processing/output : file will be /opt/ml/processing/output/test_0.csv:: \u001b[0m\n",
      "\u001b[34mREAD:Data:len=4::  shape=(4, 10)::\u001b[0m\n",
      "\u001b[34mTemp directory=/opt/ml/processing/input/filesProc:: deleted\u001b[0m\n",
      "\u001b[34mAll Done !! written out !!\u001b[0m\n",
      "\u001b[35mPandas:version:1.1.3\u001b[0m\n",
      "\u001b[35mNumpy:version:1.21.0\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess_tar.py:54: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  'SHARD_PREFIX': np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess_tar.py:55: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"YEAR_BUILT\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess_tar.py:57: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"NUM_BEDROOMS\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess_tar.py:58: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"NUM_BATHROOMS\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess_tar.py:60: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"GARAGE_SPACES\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess_tar.py:61: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"FRONT_PORCH\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distprocess_tar.py:62: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"DECK\": np.int,\u001b[0m\n",
      "\u001b[35mStarting distprocess.py.\u001b[0m\n",
      "\u001b[35mTAR:logging:all:environment:variables=data_size::value=100::\u001b[0m\n",
      "\u001b[35mprinting the env vars now\u001b[0m\n",
      "\u001b[35mTAR:logging:all:environment:variables=input_path::value=/opt/ml/processing/input::\u001b[0m\n",
      "\u001b[35m{'AWS_CONTAINER_CREDENTIALS_RELATIVE_URI': '/v2/credentials/RAbFHIwajO4F6oxZpQjR2NNzXdAZu_M2FI5Ig_UGvhc',\n",
      " 'AWS_REGION': 'us-east-1',\n",
      " 'HOME': '/root',\n",
      " 'HOSTNAME': 'ip-10-0-139-91.ec2.internal',\n",
      " 'LANG': 'C.UTF-8',\n",
      " 'LC_ALL': 'C.UTF-8',\n",
      " 'PATH': '/miniconda3/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin',\n",
      " 'PYTHONDONTWRITEBYTECODE': '1',\n",
      " 'PYTHONIOENCODING': 'UTF-8',\n",
      " 'PYTHONUNBUFFERED': '1',\n",
      " 'SAGEMAKER_SERVING_MODULE': 'sagemaker_sklearn_container.serving:main',\n",
      " 'SAGEMAKER_SKLEARN_VERSION': '0.23-1',\n",
      " 'SAGEMAKER_TRAINING_MODULE': 'sagemaker_sklearn_container.training:main',\n",
      " 'SKLEARN_MMS_CONFIG': '/home/model-server/config.properties',\n",
      " 'SM_CHECKPOINT_CONFIG_FILE': '/opt/ml/input/config/checkpointconfig.json',\n",
      " 'SM_INPUT': '/opt/ml/input',\n",
      " 'SM_INPUT_DATA_CONFIG_FILE': '/opt/ml/input/config/inputdataconfig.json',\n",
      " 'SM_INPUT_TRAINING_CONFIG_FILE': '/opt/ml/input/config/hyperparameters.json',\n",
      " 'SM_MODEL_DIR': '/opt/ml/model',\n",
      " 'TEMP': '/home/model-server/tmp'}\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input\u001b[0m\n",
      "\u001b[35mInput:data:path:=/opt/ml/processing/input::\u001b[0m\n",
      "\u001b[35mTAR:Data Downloaded::Now Reading downloaded data.:dir:/opt/ml/processing/input::And:FILES:ARE::['test_3.tar.gz', 'test_1.tar.gz']\u001b[0m\n",
      "\u001b[35mTAR:Creating directory:actual Procesing dir=/opt/ml/processing/input/filesProc:file_loc=/opt/ml/processing/input/test_3.tar.gz:\u001b[0m\n",
      "\u001b[35mTAR FILE Extracted successfully to dir=/opt/ml/processing/input/filesProc\u001b[0m\n",
      "\u001b[35mTAR FILE listing :/opt/ml/processing/input/filesProc: gives us :['data']:\u001b[0m\n",
      "\u001b[35mTAR FILE :proc_data_dir:to:/opt/ml/processing/input/filesProc/data::\u001b[0m\n",
      "\u001b[35mTAR: After extraction:READ::dir:/opt/ml/processing/input/filesProc/data::And:FILES:ARE::['test_3.csv']\u001b[0m\n",
      "\u001b[35mTAR: After extraction:FILE:loc=/opt/ml/processing/input/filesProc/data/test_3.csv::\u001b[0m\n",
      "\u001b[35mDF:SHARD:INDEX=['SHARD_PREFIX', '3', '3', '3', '3']::\u001b[0m\n",
      "\u001b[35mAFTER:READ:Going to write it to /opt/ml/processing/output : file will be /opt/ml/processing/output/test_3.csv:: \u001b[0m\n",
      "\u001b[35mREAD:Data:len=5::  shape=(5, 10)::\u001b[0m\n",
      "\u001b[35mTemp directory=/opt/ml/processing/input/filesProc:: deleted\u001b[0m\n",
      "\u001b[35mTAR:Creating directory:actual Procesing dir=/opt/ml/processing/input/filesProc:file_loc=/opt/ml/processing/input/test_1.tar.gz:\u001b[0m\n",
      "\u001b[35mTAR FILE Extracted successfully to dir=/opt/ml/processing/input/filesProc\u001b[0m\n",
      "\u001b[35mTAR FILE listing :/opt/ml/processing/input/filesProc: gives us :['data']:\u001b[0m\n",
      "\u001b[35mTAR FILE :proc_data_dir:to:/opt/ml/processing/input/filesProc/data::\u001b[0m\n",
      "\u001b[35mTAR: After extraction:READ::dir:/opt/ml/processing/input/filesProc/data::And:FILES:ARE::['test_1.csv']\u001b[0m\n",
      "\u001b[35mTAR: After extraction:FILE:loc=/opt/ml/processing/input/filesProc/data/test_1.csv::\u001b[0m\n",
      "\u001b[35mDF:SHARD:INDEX=['SHARD_PREFIX', '1', '1']::\u001b[0m\n",
      "\u001b[35mAFTER:READ:Going to write it to /opt/ml/processing/output : file will be /opt/ml/processing/output/test_1.csv:: \u001b[0m\n",
      "\u001b[35mREAD:Data:len=3::  shape=(3, 10)::\u001b[0m\n",
      "\u001b[35mTemp directory=/opt/ml/processing/input/filesProc:: deleted\u001b[0m\n",
      "\u001b[35mAll Done !! written out !!\u001b[0m\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from sagemaker.sklearn.processing import SKLearnProcessor\n",
    "from sagemaker.workflow.parameters import (\n",
    "    ParameterInteger,\n",
    "    ParameterString,\n",
    ")\n",
    "\n",
    "\n",
    "print(f\"Using:role={role}:\")\n",
    "print(f\"using SageMaker session={sm_session}:\")\n",
    "\n",
    "BASE_JOB_PREFIX=\"smjobs-dist\",  # Choose any name\n",
    "\n",
    "processing_instance_type = ParameterString(\n",
    "        name=\"ProcessingInstanceType\", default_value=\"ml.m5.xlarge\"\n",
    ")\n",
    "\n",
    "\n",
    "# Processing step for feature engineering\n",
    "sklearn_processor = SKLearnProcessor(\n",
    "        framework_version=\"0.23-1\",\n",
    "        instance_type=processing_instance_type,\n",
    "        instance_count=2,\n",
    "        base_job_name=\"dist-proc\" ,\n",
    "        sagemaker_session=sm_session,\n",
    "        role=role,\n",
    "    )\n",
    "\n",
    "from sagemaker.processing import ProcessingInput, ProcessingOutput\n",
    "inputs_p=[\n",
    "    ProcessingInput(\n",
    "        #source=\"s3://sagemaker-crossaccnt-train/data/finance/distrib\",\n",
    "        source=\"s3://sagemaker-crossaccnt-train/data/finance/distrib-multi-tar/shard\",\n",
    "        destination='/opt/ml/processing/input',\n",
    "        s3_data_distribution_type='ShardedByS3Key'\n",
    "    ),\n",
    " ]\n",
    "outputs_p=[\n",
    "    ProcessingOutput(\n",
    "        s3_upload_mode=\"EndOfJob\",\n",
    "        output_name='final_output',\n",
    "        source='/opt/ml/processing/output',\n",
    "        #destination='s3://sagemaker-crossaccnt-train/data/finance/distrib/output'\n",
    "        destination='s3://sagemaker-crossaccnt-train/data/finance/distrib-multi-tar/output'\n",
    "    ),\n",
    "    \n",
    "    \n",
    "]\n",
    "# -- if we d onot create a output then this directory is never creatd on tbe processing job\n",
    "\n",
    "preproc_script =  \"./scripts/distprocess_tar.py\"\n",
    "\n",
    "print(f\"Preprocessing:script:loc={preproc_script}:\")\n",
    "job_arguments=[\"--input-path\", \"/opt/ml/processing/input\", \n",
    "              \"--data-size\", \"100\"]\n",
    "sklearn_processor.run(\n",
    "    code=preproc_script,\n",
    "    inputs=inputs_p,\n",
    "    outputs=outputs_p,\n",
    "    arguments=job_arguments, \n",
    "    wait=True,\n",
    "    logs=\"All\",\n",
    ")\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Manifest\n",
    "Run as manifest file\n",
    "The files will be copied to:\n",
    "\n",
    "    local_path/bucket/key1\n",
    "    local_path/bucket/key2\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing ./scripts/distmanifestprocess.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile ./scripts/distmanifestprocess.py\n",
    "\n",
    "\"\"\"Feature engineers the distributed data set \"\"\"\n",
    "import argparse\n",
    "import logging\n",
    "import os\n",
    "import pathlib\n",
    "import requests\n",
    "import tempfile\n",
    "\n",
    "import boto3\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "import json\n",
    "\n",
    "import multiprocessing\n",
    "\n",
    "_logger = logging.getLogger()\n",
    "_logger.setLevel(logging.INFO)\n",
    "_logger.addHandler(logging.StreamHandler())\n",
    "\n",
    "try:\n",
    "    _logger.info(f\"Pandas:version:{pd.__version__}\")\n",
    "    _logger.info(f\"Numpy:version:{np.__version__}\")\n",
    "except:\n",
    "    pass\n",
    "\n",
    "\n",
    "# Since we get a headerless CSV file we specify the column names here.\n",
    "\n",
    "# Since we get a headerless CSV file we specify the column names here.\n",
    "\n",
    "columns=[\n",
    "    \"SHARD_PREFIX\",\n",
    "    \"YEAR_BUILT\",\n",
    "    \"SQUARE_FEET\",\n",
    "    \"NUM_BEDROOMS\",\n",
    "    \"NUM_BATHROOMS\",\n",
    "    \"LOT_ACRES\",\n",
    "    \"GARAGE_SPACES\",\n",
    "    \"FRONT_PORCH\",\n",
    "    \"DECK\",\n",
    "    \"PRICE\"\n",
    "]\n",
    "\n",
    "\n",
    "columns_dtype = {\n",
    "    'SHARD_PREFIX': np.int, \n",
    "    \"YEAR_BUILT\": np.int,\n",
    "    \"SQUARE_FEET\": np.float64,\n",
    "    \"NUM_BEDROOMS\": np.int,\n",
    "    \"NUM_BATHROOMS\": np.int,\n",
    "    \"LOT_ACRES\": np.float64,\n",
    "    \"GARAGE_SPACES\": np.int,\n",
    "    \"FRONT_PORCH\": np.int,\n",
    "    \"DECK\": np.int,\n",
    "    \"PRICE\": np.float64\n",
    "    #   'body': str, \n",
    "}\n",
    "\n",
    "def merge_two_dicts(x, y):\n",
    "    \"\"\"Merges two dicts, returning a new copy.\"\"\"\n",
    "    z = x.copy()\n",
    "    z.update(y)\n",
    "    return z\n",
    "\n",
    "def listLocalDirectory(dirPath=\".\"):\n",
    "    for path, dnames, fnames in os.walk(dirPath):\n",
    "        _logger.info(f\"List::path={path}::dirNames={dnames}::fileNames={fnames}::\")\n",
    "\n",
    "    print(\"list:done:\")\n",
    "       \n",
    "if __name__ == \"__main__\":\n",
    "    _logger.info(\"Starting distprocess.py.\")\n",
    "    parser = argparse.ArgumentParser()\n",
    "    parser.add_argument(\"--data-size\", type=int, default=100)\n",
    "    parser.add_argument(\"--input-path\", type=str, default='/opt/ml/processing/input') # \"input_path\", \"/opt/ml/processing/input\",\n",
    "    args = parser.parse_args()\n",
    "    data_size = args.data_size\n",
    "    input_dir = args.input_path\n",
    "    \n",
    "    import os\n",
    "\n",
    "    for name, value in os.environ.items():\n",
    "        _logger.info(f\"Manifest:Env:variables:name={name}::value={value}::\")\n",
    "        \n",
    "    # Split list of files into sub-lists\n",
    "    cpu_count = multiprocessing.cpu_count()\n",
    "\n",
    "    \n",
    "    BASE_DIR = \"/opt/ml/processing\"\n",
    "    \n",
    "    print(input_dir)\n",
    "    _logger.info(f\"Input:data:path:={input_dir}::\")\n",
    "\n",
    "    \n",
    "    onlyFiles = [f for f in os.listdir(input_dir) if os.path.isfile(os.path.join(input_dir, f))]\n",
    "    _logger.info(f\"Data Downloaded::Now Reading downloaded data.:dir:{input_dir}::And:FILES:ARE::{onlyFiles}\")\n",
    "    \n",
    "    files = []\n",
    "    for r, d, f in os.walk(input_dir):\n",
    "        for file in f:\n",
    "            files.append(os.path.join(r, file))\n",
    "            _logger.info(f\"Manifest:r={r}::d={d}::f={f}::file_loc={os.path.join(r, file)}::\")\n",
    "\n",
    "    \n",
    "    fn = os.path.join(input_dir, onlyFiles[0])\n",
    "    \n",
    "    # read in csv\n",
    "    df = pd.read_csv(fn, names=columns)\n",
    "    #_logger.info(f\"df:head={df.head()}\")\n",
    "    _logger.info(f\"DF:SHARD:INDEX={df['SHARD_PREFIX'].to_list()}::\")\n",
    "    \n",
    "    _logger.info(f\"AFTER:READ:Going to write it to {BASE_DIR}/output : file will be {BASE_DIR}/output/{onlyFiles[0]}:: \")\n",
    "    _logger.info(f\"READ:Data:len={len(df)}::  shape={df.shape}::\")\n",
    "    df.to_csv(\n",
    "        f\"{BASE_DIR}/output/{onlyFiles[0]}\", header=True, index=False\n",
    "    )\n",
    "    \n",
    "    _logger.info(\"All Done !! written out !!\")\n",
    "    \n",
    "    # ----------------   OLD TEMPLATE CODE --------------------#"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 139,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting ./scripts/manifest.txt\n"
     ]
    }
   ],
   "source": [
    "%%writefile ./scripts/manifest.txt\n",
    "[ \n",
    "    {\n",
    "        \"prefix\": \"s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard/\"\n",
    "    },\n",
    "    \"sh0/test_0.csv\",\n",
    "    \"sh0/text_file_0.txt\",\n",
    "    \"sh1/test_1.csv\",\n",
    "    \"sh2/test_2.csv\",\n",
    "    \"sh3/test_3.csv\"\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 140,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard/manifest.txt'"
      ]
     },
     "execution_count": 140,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "s3_upload_uri = sagemaker.s3.S3Uploader.upload(\n",
    "            local_path='./scripts/manifest.txt',\n",
    "            desired_s3_uri=f\"s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard\",\n",
    "            kms_key=None,\n",
    "            sagemaker_session=sm_session\n",
    "        )\n",
    "s3_upload_uri"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Results\n",
    "    *Using  1 file at the prefix location using Fully Replicated works\n",
    "    *Using 2 files at the prefix location using Fully Replicated works\n",
    "\n",
    "    *Using 1 file but using shardbys3 at the prefix location works\n",
    "    *Using 2 file but using ShardByS3 at the prefix location works\n",
    "\n",
    "    *Using 1 file but nested from prefix location using Fully Replicated -- not tested\n",
    "    *Using 1 file but nested from prefix location using SHARDByS3 works\n",
    "    \n",
    "    *Using 2 file but nested from prefix location using Fully Replicated -- not tested\n",
    "    *Using 2 file but nested from prefix location using SHARDByS3 works\n",
    "    \n",
    "    *Using 4 file but uploaded using S3 uploader and nested from prefix location using SHARDByS3 - works\n",
    "    \n",
    "    *Using 4 files but one of them is a folder and nested from prefix location using ShardByS3 -- Fails and does not work\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 141,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Using:role=arn:aws:iam::622343165275:role/service-role/AmazonSageMaker-ExecutionRole-20220208T115633:\n",
      "using SageMaker session=<sagemaker.session.Session object at 0x7f5ef5bbf790>:\n",
      "Preprocessing:script:loc=./scripts/distmanifestprocess.py:\n",
      "\n",
      "Job Name:  dist-proc-2022-06-08-22-59-51-235\n",
      "Inputs:  [{'InputName': 'ManifestFileInput', 'AppManaged': False, 'S3Input': {'S3Uri': 's3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard/manifest.txt', 'LocalPath': '/opt/ml/processing/input', 'S3DataType': 'ManifestFile', 'S3InputMode': 'File', 'S3DataDistributionType': 'ShardedByS3Key'}}, {'InputName': 'code', 'AppManaged': False, 'S3Input': {'S3Uri': 's3://sagemaker-crossaccnt-train/dist-proc-2022-06-08-22-59-51-235/input/code/distmanifestprocess.py', 'LocalPath': '/opt/ml/processing/input/code', 'S3DataType': 'S3Prefix', 'S3InputMode': 'File', 'S3DataDistributionType': 'FullyReplicated', 'S3CompressionType': 'None'}}]\n",
      "Outputs:  [{'OutputName': 'final_output', 'AppManaged': False, 'S3Output': {'S3Uri': 's3://sagemaker-crossaccnt-train/data/finance/distrib-multi/output', 'LocalPath': '/opt/ml/processing/output', 'S3UploadMode': 'EndOfJob'}}]\n",
      "...............................\u001b[34mPandas:version:1.1.3\u001b[0m\n",
      "\u001b[34mNumpy:version:1.21.0\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distmanifestprocess.py:48: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  'SHARD_PREFIX': np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distmanifestprocess.py:49: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"YEAR_BUILT\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distmanifestprocess.py:51: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"NUM_BEDROOMS\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distmanifestprocess.py:52: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"NUM_BATHROOMS\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distmanifestprocess.py:54: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"GARAGE_SPACES\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distmanifestprocess.py:55: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"FRONT_PORCH\": np.int,\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input/code/distmanifestprocess.py:56: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[34mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"DECK\": np.int,\u001b[0m\n",
      "\u001b[34mStarting distprocess.py.\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input\u001b[0m\n",
      "\u001b[34mInput:data:path:=/opt/ml/processing/input::\u001b[0m\n",
      "\u001b[34mData Downloaded::Now Reading downloaded data.:dir:/opt/ml/processing/input::And:FILES:ARE::[]\u001b[0m\n",
      "\u001b[34mManifest:r=/opt/ml/processing/input/sh3::d=[]::f=['test_3.csv']::file_loc=/opt/ml/processing/input/sh3/test_3.csv::\u001b[0m\n",
      "\u001b[34mManifest:r=/opt/ml/processing/input/code::d=[]::f=['distmanifestprocess.py']::file_loc=/opt/ml/processing/input/code/distmanifestprocess.py::\u001b[0m\n",
      "\u001b[34mManifest:r=/opt/ml/processing/input/sh0::d=[]::f=['test_0.csv']::file_loc=/opt/ml/processing/input/sh0/test_0.csv::\u001b[0m\n",
      "\u001b[34mManifest:r=/opt/ml/processing/input/sh1::d=[]::f=['test_1.csv']::file_loc=/opt/ml/processing/input/sh1/test_1.csv::\u001b[0m\n",
      "\u001b[34mTraceback (most recent call last):\n",
      "  File \"/opt/ml/processing/input/code/distmanifestprocess.py\", line 102, in <module>\n",
      "    fn = os.path.join(input_dir, onlyFiles[0])\u001b[0m\n",
      "\u001b[34mIndexError: list index out of range\u001b[0m\n",
      "\u001b[35mPandas:version:1.1.3\u001b[0m\n",
      "\u001b[35mNumpy:version:1.21.0\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distmanifestprocess.py:48: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  'SHARD_PREFIX': np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distmanifestprocess.py:49: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"YEAR_BUILT\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distmanifestprocess.py:51: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"NUM_BEDROOMS\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distmanifestprocess.py:52: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"NUM_BATHROOMS\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distmanifestprocess.py:54: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"GARAGE_SPACES\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distmanifestprocess.py:55: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"FRONT_PORCH\": np.int,\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input/code/distmanifestprocess.py:56: DeprecationWarning: `np.int` is a deprecated alias for the builtin `int`. To silence this warning, use `int` by itself. Doing this will not modify any behavior and is safe. When replacing `np.int`, you may wish to use e.g. `np.int64` or `np.int32` to specify the precision. If you wish to review your current use, check the release note link for additional information.\u001b[0m\n",
      "\u001b[35mDeprecated in NumPy 1.20; for more details and guidance: https://numpy.org/devdocs/release/1.20.0-notes.html#deprecations\n",
      "  \"DECK\": np.int,\u001b[0m\n",
      "\u001b[35mStarting distprocess.py.\u001b[0m\n",
      "\u001b[35m/opt/ml/processing/input\u001b[0m\n",
      "\u001b[35mInput:data:path:=/opt/ml/processing/input::\u001b[0m\n",
      "\u001b[35mData Downloaded::Now Reading downloaded data.:dir:/opt/ml/processing/input::And:FILES:ARE::[]\u001b[0m\n",
      "\u001b[35mManifest:r=/opt/ml/processing/input/sh0::d=[]::f=['text_file_0.txt']::file_loc=/opt/ml/processing/input/sh0/text_file_0.txt::\u001b[0m\n",
      "\u001b[35mManifest:r=/opt/ml/processing/input/code::d=[]::f=['distmanifestprocess.py']::file_loc=/opt/ml/processing/input/code/distmanifestprocess.py::\u001b[0m\n",
      "\u001b[35mManifest:r=/opt/ml/processing/input/sh2::d=[]::f=['test_2.csv']::file_loc=/opt/ml/processing/input/sh2/test_2.csv::\u001b[0m\n",
      "\u001b[35mTraceback (most recent call last):\n",
      "  File \"/opt/ml/processing/input/code/distmanifestprocess.py\", line 102, in <module>\n",
      "    fn = os.path.join(input_dir, onlyFiles[0])\u001b[0m\n",
      "\u001b[35mIndexError: list index out of range\u001b[0m\n",
      "\n"
     ]
    },
    {
     "ename": "UnexpectedStatusException",
     "evalue": "Error for Processing job dist-proc-2022-06-08-22-59-51-235: Failed. Reason: AlgorithmError: See job logs for more information",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mUnexpectedStatusException\u001b[0m                 Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-141-2449d4e3af7e>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m     62\u001b[0m     \u001b[0marguments\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mjob_arguments\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     63\u001b[0m     \u001b[0mwait\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mTrue\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 64\u001b[0;31m     \u001b[0mlogs\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m\"All\"\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     65\u001b[0m )\n",
      "\u001b[0;32m/opt/conda/lib/python3.7/site-packages/sagemaker/processing.py\u001b[0m in \u001b[0;36mrun\u001b[0;34m(self, code, inputs, outputs, arguments, wait, logs, job_name, experiment_config, kms_key)\u001b[0m\n\u001b[1;32m    551\u001b[0m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjobs\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mappend\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mlatest_job\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    552\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mwait\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 553\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mlatest_job\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mwait\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mlogs\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mlogs\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    554\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    555\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0m_include_code_in_inputs\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0minputs\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mcode\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mkms_key\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mNone\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/opt/conda/lib/python3.7/site-packages/sagemaker/processing.py\u001b[0m in \u001b[0;36mwait\u001b[0;34m(self, logs)\u001b[0m\n\u001b[1;32m    960\u001b[0m         \"\"\"\n\u001b[1;32m    961\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mlogs\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 962\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msagemaker_session\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mlogs_for_processing_job\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjob_name\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mwait\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mTrue\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    963\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    964\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msagemaker_session\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mwait_for_processing_job\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjob_name\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/opt/conda/lib/python3.7/site-packages/sagemaker/session.py\u001b[0m in \u001b[0;36mlogs_for_processing_job\u001b[0;34m(self, job_name, wait, poll)\u001b[0m\n\u001b[1;32m   3879\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   3880\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mwait\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 3881\u001b[0;31m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_check_job_status\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mjob_name\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mdescription\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m\"ProcessingJobStatus\"\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   3882\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0mdot\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   3883\u001b[0m                 \u001b[0mprint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m/opt/conda/lib/python3.7/site-packages/sagemaker/session.py\u001b[0m in \u001b[0;36m_check_job_status\u001b[0;34m(self, job, desc, status_key_name)\u001b[0m\n\u001b[1;32m   3331\u001b[0m                 \u001b[0mmessage\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mmessage\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   3332\u001b[0m                 \u001b[0mallowed_statuses\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m\"Completed\"\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m\"Stopped\"\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 3333\u001b[0;31m                 \u001b[0mactual_status\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mstatus\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   3334\u001b[0m             )\n\u001b[1;32m   3335\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mUnexpectedStatusException\u001b[0m: Error for Processing job dist-proc-2022-06-08-22-59-51-235: Failed. Reason: AlgorithmError: See job logs for more information"
     ]
    }
   ],
   "source": [
    "from sagemaker.sklearn.processing import SKLearnProcessor\n",
    "from sagemaker.workflow.parameters import (\n",
    "    ParameterInteger,\n",
    "    ParameterString,\n",
    ")\n",
    "\n",
    "print(f\"Using:role={role}:\")\n",
    "print(f\"using SageMaker session={sm_session}:\")\n",
    "\n",
    "BASE_JOB_PREFIX=\"smjobs-dist\",  # Choose any name\n",
    "\n",
    "processing_instance_type = ParameterString(\n",
    "        name=\"ProcessingInstanceType\", default_value=\"ml.m5.xlarge\"\n",
    ")\n",
    "# Processing step for feature engineering\n",
    "sklearn_processor = SKLearnProcessor(\n",
    "        framework_version=\"0.23-1\",\n",
    "        instance_type=processing_instance_type,\n",
    "        instance_count=2,\n",
    "        base_job_name=\"dist-proc\" ,\n",
    "        sagemaker_session=sm_session,\n",
    "        role=role,\n",
    "    )\n",
    "\n",
    "from sagemaker.processing import ProcessingInput, ProcessingOutput, S3Input\n",
    "\n",
    "inputs_p=[\n",
    "    ProcessingInput(\n",
    "        input_name=\"ManifestFileInput\",\n",
    "        s3_input=S3Input(\n",
    "            s3_uri=\"s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard/manifest.txt\",\n",
    "            local_path=\"/opt/ml/processing/input\",\n",
    "            s3_data_type=\"ManifestFile\",\n",
    "            s3_input_mode='File',\n",
    "            s3_data_distribution_type='ShardedByS3Key',#'ShardedByS3Key', 'FullyReplicated'\n",
    "        )\n",
    "    ),\n",
    "]\n",
    "\n",
    "outputs_p=[\n",
    "    ProcessingOutput(\n",
    "        s3_upload_mode=\"EndOfJob\",\n",
    "        output_name='final_output',\n",
    "        source='/opt/ml/processing/output',\n",
    "        #destination='s3://sagemaker-crossaccnt-train/data/finance/distrib/output'\n",
    "        destination='s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/output'\n",
    "    ),\n",
    "    \n",
    "    \n",
    "]\n",
    "# -- if we d onot create a output then this directory is never creatd on tbe processing job\n",
    "\n",
    "preproc_script =  \"./scripts/distmanifestprocess.py\"\n",
    "\n",
    "print(f\"Preprocessing:script:loc={preproc_script}:\")\n",
    "job_arguments=[\"--input-path\", \"/opt/ml/processing/input\", \n",
    "              \"--data-size\", \"100\"]\n",
    "sklearn_processor.run(\n",
    "    code=preproc_script,\n",
    "    inputs=inputs_p,\n",
    "    outputs=outputs_p,\n",
    "    arguments=job_arguments, \n",
    "    wait=True,\n",
    "    logs=\"All\",\n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "instance_type": "ml.m5.2xlarge",
  "kernelspec": {
   "display_name": "Python 3 (Data Science)",
   "language": "python",
   "name": "python3__SAGEMAKER_INTERNAL__arn:aws:sagemaker:us-east-1:081325390199:image/datascience-1.0"
  },
  "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.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}