TarBallShard.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "sagemaker:version:2.83.0\n",
      "Pandas:version:1.3.5\n",
      "Numpy:version:1.21.4\n"
     ]
    }
   ],
   "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",
    "\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 warnings\n",
    "warnings.filterwarnings('ignore')\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "DEF_BUCKET=\"sagemaker-crossaccnt-train\"\n",
    "PREFIX=\"data/finance/distrib-multi\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'sagemaker-crossaccnt-train'"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "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=DEF_BUCKET)\n",
    "default_bucket = sm_session.default_bucket()\n",
    "sm_client=sm_session.sagemaker_client\n",
    "sm_session.default_bucket()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Generate Data\n",
    "* Generate synthetic housing data -- with no Categorical columns for now -- \n",
    "    * SHARD is RANDOM INDEX and so everytime we run we get a new set of shard indexes\n",
    "    * Generate Json File -- not included as yet\n",
    "    * Generate Zip file - not included as yet"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "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": 10,
   "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>2002</td>\n",
       "      <td>2587.245611</td>\n",
       "      <td>2</td>\n",
       "      <td>3.0</td>\n",
       "      <td>1.65</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>407836</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>0</td>\n",
       "      <td>2004</td>\n",
       "      <td>3932.424543</td>\n",
       "      <td>2</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.30</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>599363</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>1</td>\n",
       "      <td>2002</td>\n",
       "      <td>2110.358082</td>\n",
       "      <td>6</td>\n",
       "      <td>2.5</td>\n",
       "      <td>1.02</td>\n",
       "      <td>2</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>374353</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2</td>\n",
       "      <td>2009</td>\n",
       "      <td>2688.848561</td>\n",
       "      <td>2</td>\n",
       "      <td>2.0</td>\n",
       "      <td>0.84</td>\n",
       "      <td>2</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>445927</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>1</td>\n",
       "      <td>1995</td>\n",
       "      <td>2356.615287</td>\n",
       "      <td>2</td>\n",
       "      <td>2.0</td>\n",
       "      <td>1.41</td>\n",
       "      <td>3</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>349642</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>0</td>\n",
       "      <td>1987</td>\n",
       "      <td>2806.100714</td>\n",
       "      <td>3</td>\n",
       "      <td>1.5</td>\n",
       "      <td>0.69</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>338765</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>1</td>\n",
       "      <td>1986</td>\n",
       "      <td>3359.818625</td>\n",
       "      <td>3</td>\n",
       "      <td>1.0</td>\n",
       "      <td>1.18</td>\n",
       "      <td>3</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>446672</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>2</td>\n",
       "      <td>2002</td>\n",
       "      <td>3482.570149</td>\n",
       "      <td>4</td>\n",
       "      <td>3.0</td>\n",
       "      <td>1.26</td>\n",
       "      <td>3</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "      <td>586285</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>3</td>\n",
       "      <td>2000</td>\n",
       "      <td>2275.129275</td>\n",
       "      <td>6</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.40</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>327269</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>2</td>\n",
       "      <td>1995</td>\n",
       "      <td>2868.584367</td>\n",
       "      <td>4</td>\n",
       "      <td>3.0</td>\n",
       "      <td>0.91</td>\n",
       "      <td>1</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>423937</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   SHARD_PREFIX  YEAR_BUILT  SQUARE_FEET  NUM_BEDROOMS  NUM_BATHROOMS  \\\n",
       "0             3        2002  2587.245611             2            3.0   \n",
       "1             0        2004  3932.424543             2            2.0   \n",
       "2             1        2002  2110.358082             6            2.5   \n",
       "3             2        2009  2688.848561             2            2.0   \n",
       "4             1        1995  2356.615287             2            2.0   \n",
       "5             0        1987  2806.100714             3            1.5   \n",
       "6             1        1986  3359.818625             3            1.0   \n",
       "7             2        2002  3482.570149             4            3.0   \n",
       "8             3        2000  2275.129275             6            1.0   \n",
       "9             2        1995  2868.584367             4            3.0   \n",
       "\n",
       "   LOT_ACRES  GARAGE_SPACES  FRONT_PORCH  DECK   PRICE  \n",
       "0       1.65              1            0     0  407836  \n",
       "1       1.30              1            1     1  599363  \n",
       "2       1.02              2            1     1  374353  \n",
       "3       0.84              2            1     1  445927  \n",
       "4       1.41              3            0     1  349642  \n",
       "5       0.69              1            0     0  338765  \n",
       "6       1.18              3            0     0  446672  \n",
       "7       1.26              3            1     1  586285  \n",
       "8       0.40              0            0     0  327269  \n",
       "9       0.91              1            0     0  423937  "
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = generate_houses(10)\n",
    "df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[3, 0, 1, 2, 1, 0, 1, 2, 3, 2]\n"
     ]
    }
   ],
   "source": [
    "print(df['SHARD_PREFIX'].to_list())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "S3:::data_loc=./data/test_0.csv::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard/sh0/test_0.csv\n",
      "S3:::data_loc=./data/test_1.csv::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard/sh1/test_1.csv\n",
      "S3:::data_loc=./data/test_2.csv::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/shard/sh2/test_2.csv\n",
      "S3:::data_loc=./data/test_3.csv::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://{DEF_BUCKET}/{PREFIX}/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:::data_loc={data_loc}::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": 14,
   "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/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/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/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/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",
    "        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://{DEF_BUCKET}/{PREFIX}/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": "markdown",
   "metadata": {},
   "source": [
    "## Write Tar\n",
    "tar processessing script"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "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": 19,
   "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 Tar\n",
    "Run the Tar processing job"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "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 0x7fb0fcda7710>:\n",
      "Preprocessing:script:loc=./scripts/distprocess_tar.py:\n",
      "\n",
      "Job Name:  dist-proc-2022-06-09-22-44-07-808\n",
      "Inputs:  [{'InputName': 'input-1', 'AppManaged': False, 'S3Input': {'S3Uri': 's3://sagemaker-crossaccnt-train/data/finance/distrib-multi/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-09-22-44-07-808/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/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/ngSZkUO52tRuc6KRpKUN2zGSfww7hwKut_kNiMyVSk4',\n",
      " 'AWS_REGION': 'us-east-1',\n",
      " 'HOME': '/root',\n",
      " 'HOSTNAME': 'ip-10-0-110-232.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',\u001b[0m\n",
      "\u001b[34mInput:data:path:=/opt/ml/processing/input::\n",
      " 'TEMP': '/home/model-server/tmp'}\u001b[0m\n",
      "\u001b[34m/opt/ml/processing/input\u001b[0m\n",
      "\u001b[34merror in reading input data config\u001b[0m\n",
      "\u001b[34mTAR: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[34mTAR:Creating directory:actual Procesing dir=/opt/ml/processing/input/filesProc:file_loc=/opt/ml/processing/input/test_3.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_3.csv']\u001b[0m\n",
      "\u001b[34mTAR: After extraction:FILE:loc=/opt/ml/processing/input/filesProc/data/test_3.csv::\u001b[0m\n",
      "\u001b[34mDF:SHARD:INDEX=['SHARD_PREFIX', '3', '3']::\u001b[0m\n",
      "\u001b[34mAFTER:READ:Going to write it to /opt/ml/processing/output : file will be /opt/ml/processing/output/test_3.csv:: \u001b[0m\n",
      "\u001b[34mREAD:Data:len=3::  shape=(3, 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_1.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_1.csv']\u001b[0m\n",
      "\u001b[34mTAR: After extraction:FILE:loc=/opt/ml/processing/input/filesProc/data/test_1.csv::\u001b[0m\n",
      "\u001b[34mDF:SHARD:INDEX=['SHARD_PREFIX', '1', '1', '1']::\u001b[0m\n",
      "\u001b[34mAFTER:READ:Going to write it to /opt/ml/processing/output : file will be /opt/ml/processing/output/test_1.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[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/ngSZkUO52tRuc6KRpKUN2zGSfww7hwKut_kNiMyVSk4',\n",
      " 'AWS_REGION': 'us-east-1',\n",
      " 'HOME': '/root',\n",
      " 'HOSTNAME': 'ip-10-0-84-216.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[34merror in reading input data config\u001b[0m\n",
      "\u001b[34mTAR:Data Downloaded::Now Reading downloaded data.:dir:/opt/ml/processing/input::And:FILES:ARE::['test_0.tar.gz', 'test_2.tar.gz']\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']::\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=3::  shape=(3, 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_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', '2', '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=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",
      "\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=f\"s3://{DEF_BUCKET}/{PREFIX}/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=f's3://{DEF_BUCKET}/{PREFIX}/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": []
  }
 ],
 "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
}