DirectFileShard.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": "raw",
   "metadata": {},
   "source": [
    "S3:::data_loc=./data/test_0.csv::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/directshard/test_0.csv\n",
    "S3:::data_loc=./data/test_1.csv::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/directshard/test_1.csv\n",
    "S3:::data_loc=./data/test_2.csv::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/directshard/test_2.csv\n",
    "S3:::data_loc=./data/test_3.csv::Uploaded:=s3://sagemaker-crossaccnt-train/data/finance/distrib-multi/directshard/test_3.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "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}/directshard\"\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": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing ./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": "code",
   "execution_count": 8,
   "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": 10,
   "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 0x7fb854d983d0>:\n",
      "Preprocessing:script:loc=./scripts/distprocess.py:\n",
      "\n",
      "Job Name:  dist-proc-2022-06-09-22-52-19-655\n",
      "Inputs:  [{'InputName': 'input-1', 'AppManaged': False, 'S3Input': {'S3Uri': 's3://sagemaker-crossaccnt-train/data/finance/distrib-multi/directshard', '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-52-19-655/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/directshard/output', 'LocalPath': '/opt/ml/processing/output', 'S3UploadMode': 'EndOfJob'}}]\n",
      "...........................\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::['test_2.csv', 'test_0.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[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.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::['test_3.csv', 'test_1.csv']\u001b[0m\n",
      "\u001b[35mDF:SHARD:INDEX=['SHARD_PREFIX', '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=3::  shape=(3, 10)::\u001b[0m\n",
      "\u001b[35mAll Done !! written out !!\u001b[0m\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}/directshard\",\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}/directshard/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": "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
}