sagemaker-geospatial/digital-farming-pipelines/pipelines-sagemaker-geospatial.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Building Pipelines with Amazon SageMaker Geospatial Capabilities\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "This notebook's CI test result for us-west-2 is as follows. CI test results in other regions can be found at the end of the notebook. \n",
    "\n",
    "![This us-west-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/us-west-2/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "\n",
    "In this notebook, we will describe an example of how to build pipelines for automating the processing of geospatial data, using Amazon SageMaker geospatial capabilities and Amazon SageMaker Pipelines.\n",
    "\n",
    "<center><img src=\"./img/pipeline.png\" width=\"80%\"></center>\n",
    "\n",
    "----"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will start by making sure the \"sagemaker\" SDK is updated, and importing a few libraries required."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true,
    "tags": []
   },
   "outputs": [],
   "source": [
    "!pip install sagemaker --upgrade"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import boto3\n",
    "import sagemaker\n",
    "import json\n",
    "from datetime import datetime"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will now define a few variables for which we need to create sessions in the SageMaker and Boto3 SDKs.\n",
    "\n",
    "We will also create the client for SageMaker geospatial capabilities with a Boto session..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "sagemaker_session = sagemaker.Session()\n",
    "bucket = sagemaker_session.default_bucket()  ### Replace with your own bucket if needed\n",
    "role = sagemaker.get_execution_role(sagemaker_session)\n",
    "sess = boto3.Session()\n",
    "region = sess.region_name\n",
    "prefix = \"sm-geospatial-e2e\"  ### Replace with the S3 prefix desired\n",
    "print(f\"S3 bucket: {bucket}\")\n",
    "print(f\"Role: {role}\")\n",
    "print(f\"Region: {region}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Before we continue make sure that your AWS IAM role has the proper permissions for interacting with AWS Lambda, Amazon S3, and Amazon SQS as required.\n",
    "\n",
    "Also, make sure you have the proper policy and trust relationship added to your role for \"sagemaker-geospatial\", as specified in the [Get Started with Amazon SageMaker Geospatial Capabiltiies](https://docs.aws.amazon.com/sagemaker/latest/dg/geospatial-getting-started.html) documentation.\n",
    "\n",
    "For demo purposes, you can uncomment and run the following cell to add the policies and trust relationships to your role using the script below, but this policy should be scoped down to improve security for any production deployment, following the least privilege principle."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "#!python './code/adjust_role.py' {role}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's now create an AWS Lambda function that will work for calling the SageMaker geospatial processes as required. We will use the same function for:\n",
    "* Starting the EOJs (Cloud Removal and Stacking in our example) - Using a Lambda Step in SageMaker Pipelines\n",
    "* Checking the status of the EOJs, as these are asynchronous and take a few minutes to complete - Using a Callback Step in SageMaker Pipelines\n",
    "\n",
    "We will start by writting a script with our code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%writefile ./code/eoj_lambda.py\n",
    "\"\"\"Script for calling SageMaker geospatial APIs as required\"\"\"\n",
    "import json\n",
    "import logging\n",
    "import os\n",
    "import boto3\n",
    "import botocore\n",
    "import ast\n",
    "import time\n",
    "\n",
    "logger = logging.getLogger()\n",
    "logger.setLevel(os.getenv(\"LOGGING_LEVEL\", logging.INFO))\n",
    "\n",
    "\n",
    "def lambda_handler(event, context):\n",
    "    \"\"\"\n",
    "    Manages SageMaker geospatial EOJs as required.\n",
    "    \"\"\"\n",
    "    try:\n",
    "        # Setup client...\n",
    "        if \"region\" in event:\n",
    "            region = event[\"region\"]\n",
    "        else:\n",
    "            region = \"us-west-2\"\n",
    "        gsClient = boto3.client(\"sagemaker-geospatial\", region_name=\"us-west-2\")\n",
    "\n",
    "        if \"eoj_name\" in event:\n",
    "            # Create a new EOJ...\n",
    "            logger.debug(f\"Will create EOJ with event:\\n{json.dumps(event)}\")\n",
    "            if \"RasterDataCollectionQuery\" in str(event[\"eoj_input_config\"]):\n",
    "                # Input is a Raster Data Collection Query\n",
    "                input_config = ast.literal_eval(event[\"eoj_input_config\"])\n",
    "            elif \"arn\" in str(event[\"eoj_input_config\"]):\n",
    "                # Input is chaining results of another EOJ\n",
    "                input_config = {\"PreviousEarthObservationJobArn\": event[\"eoj_input_config\"]}\n",
    "            logger.info(f'Starting EOJ {event[\"eoj_name\"]}')\n",
    "            response = gsClient.start_earth_observation_job(\n",
    "                Name=event[\"eoj_name\"],\n",
    "                ExecutionRoleArn=event[\"role\"],\n",
    "                InputConfig=input_config,\n",
    "                JobConfig=ast.literal_eval(event[\"eoj_config\"]),\n",
    "            )\n",
    "            logger.info(f'Create eoj_arn: {response[\"Arn\"]}\\n')\n",
    "            time.sleep(3)\n",
    "\n",
    "        elif \"eoj_output_config\" in event:\n",
    "            # Export an EOJ...\n",
    "            logger.debug(f\"Will export EOJ with event:\\n{json.dumps(event)}\")\n",
    "            logger.info(f'Exporting EOJ with Arn {event[\"eoj_arn\"]}')\n",
    "            response = gsClient.export_earth_observation_job(\n",
    "                Arn=event[\"eoj_arn\"],\n",
    "                ExecutionRoleArn=event[\"role\"],\n",
    "                OutputConfig=ast.literal_eval(event[\"eoj_output_config\"]),\n",
    "            )\n",
    "            logger.info(f'Export eoj_arn: {response[\"Arn\"]}\\n')\n",
    "\n",
    "        elif \"Records\" in event:\n",
    "            # Check status of previous EOJ...\n",
    "            logger.debug(f\"Will check status of EOJ with event:\\n{json.dumps(event)}\")\n",
    "            for record in event[\"Records\"]:\n",
    "                payload = json.loads(record[\"body\"])\n",
    "                token = payload[\"token\"]\n",
    "                eoj_arn = payload[\"arguments\"][\"eoj_arn\"]\n",
    "                logger.info(f\"Check EOJ or export with ARN: {eoj_arn}\")\n",
    "                response = gsClient.get_earth_observation_job(Arn=eoj_arn)\n",
    "                if response[\"Status\"] == \"COMPLETED\":\n",
    "                    # EOJ is COMPLETED\n",
    "                    logger.info(\"EOJ completed, resuming pipeline...\")\n",
    "                    sagemaker = boto3.client(\"sagemaker\", region_name=region)\n",
    "                    sagemaker.send_pipeline_execution_step_success(\n",
    "                        CallbackToken=token,\n",
    "                        OutputParameters=[{\"Name\": \"eoj_status\", \"Value\": response[\"Status\"]}],\n",
    "                    )\n",
    "                elif response[\"Status\"] == \"SUCCEEDED\":\n",
    "                    # Export of EOJ SUCCEEDED\n",
    "                    logger.info(\"Export EOJ succeeded, resuming pipeline...\")\n",
    "                    sagemaker = boto3.client(\"sagemaker\", region_name=region)\n",
    "                    sagemaker.send_pipeline_execution_step_sucess(\n",
    "                        CallbackToken=token,\n",
    "                        OutputParameters=[\n",
    "                            {\"Name\": \"export_eoj_status\", \"Value\": response[\"Status\"]}\n",
    "                        ],\n",
    "                    )\n",
    "                elif response[\"Status\"] == \"FAILED\":\n",
    "                    logger.info(\"EOJ or export failed, stopping pipeline...\")\n",
    "                    sagemaker = boto3.client(\"sagemaker\", region_name=region)\n",
    "                    sagemaker.send_pipeline_execution_step_failure(\n",
    "                        CallbackToken=token, FailureReason=response[\"ErrorDetails\"]\n",
    "                    )\n",
    "                else:\n",
    "                    # EOJ is still running IN_PROGRESS, we must check again later\n",
    "                    # Note we must raise an exception for having the message put back to the SNS queue\n",
    "                    logger.info(f'EOJ or export with status: {response[\"Status\"]}')\n",
    "                    raise Exception(\"EOJ or export still running...\")\n",
    "    except botocore.exceptions.ClientError as e:\n",
    "        error_msg = f\"EOJ or export call failed: {e.response['Error']['Code']}, {e.response['Error']['Message']}\"\n",
    "        raise Exception(error_msg)\n",
    "\n",
    "    try:\n",
    "        response\n",
    "    except NameError:\n",
    "        response = None\n",
    "\n",
    "    if response is not None:\n",
    "        logger.info(f'eoj_arn: {response[\"Arn\"]}\\n')\n",
    "    else:\n",
    "        response = {}\n",
    "        response[\"Arn\"] = \"\"\n",
    "\n",
    "    return {\"statusCode\": 200, \"eoj_arn\": response[\"Arn\"]}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now use this script for creating our Lambda function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Create the Lambda function...\n",
    "import zipfile\n",
    "\n",
    "with zipfile.ZipFile(\"./code/eoj_lambda.zip\", \"w\") as zf:\n",
    "    zf.write(\"./code/eoj_lambda.py\", \"eoj_lambda.py\")\n",
    "\n",
    "with open(\"./code/eoj_lambda.zip\", \"rb\") as f:\n",
    "    zipped_code = f.read()\n",
    "\n",
    "lambda_client = boto3.client(\"lambda\", region_name=region)\n",
    "\n",
    "response = lambda_client.create_function(\n",
    "    FunctionName=\"geospatial-lambda\",\n",
    "    Runtime=\"python3.9\",\n",
    "    Role=role,\n",
    "    Handler=\"eoj_lambda.lambda_handler\",\n",
    "    Code=dict(ZipFile=zipped_code),\n",
    "    Timeout=60,\n",
    "    # Set up Lambda function environment variables\n",
    "    Environment={\n",
    "        \"Variables\": {\"Name\": \"geospatial-lambda\", \"Environment\": \"prod\", \"LOGGING_LEVEL\": \"INFO\"}\n",
    "    },\n",
    ")\n",
    "\n",
    "function_arn = response[\"FunctionArn\"]\n",
    "print(f\"Created Lambda function with ARN: {function_arn}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that we have a Lambda function for calling our EOJs, we can implement the Callback steps required in our pipeline.\n",
    "\n",
    "Note again, we are using SageMaker Pipelines Callback Steps because our EOJs are asynchronous and takes sometime to complete. So we want the Lambda to check the status of the EOJs, and resume the workflows when each EOJ is completed."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For this, we will create an Amazon SQS queue that will be used in our callback."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Create SQS queue for handling async processes and callback\n",
    "sqs_client = boto3.client(\"sqs\", region_name=region)\n",
    "\n",
    "sqs_client.create_queue(\n",
    "    QueueName=f\"geospatial-queue\",\n",
    "    Attributes={\n",
    "        \"VisibilityTimeout\": \"300\",\n",
    "        \"DelaySeconds\": \"5\",\n",
    "        \"ReceiveMessageWaitTimeSeconds\": \"5\",\n",
    "    },\n",
    ")\n",
    "queue_url = sqs_client.get_queue_url(QueueName=f\"geospatial-queue\")[\"QueueUrl\"]\n",
    "queue_arn = sqs_client.get_queue_attributes(QueueUrl=queue_url, AttributeNames=[\"QueueArn\"])[\n",
    "    \"Attributes\"\n",
    "][\"QueueArn\"]\n",
    "\n",
    "print(f\"Created queue:\\n{queue_url}\\n{queue_arn}\\n\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we need to associate the SQS queue as an input trigger for our Lambda function, in this way whenever the Callback Step pushes a message to the queue it would run our Lambda function for checking the status of the EOJ. We do this by creating an Even Source Mapping."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Link SQS queue with the geospatial-check-lambda function...\n",
    "event_source_mapping = lambda_client.create_event_source_mapping(\n",
    "    EventSourceArn=queue_arn, FunctionName=\"geospatial-lambda\", Enabled=True\n",
    ")\n",
    "print(f'Mapping Lambda function and SQS queue through UUID: {event_source_mapping[\"UUID\"]}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will now define the parameters to be used in our pipeline, as we want to be able to pass these dynamically whenever we run our geospatial pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from sagemaker.workflow.parameters import ParameterString\n",
    "\n",
    "parameter_role = ParameterString(name=\"parameter_role\", default_value=role)\n",
    "parameter_region = ParameterString(name=\"parameter_region\", default_value=region)\n",
    "parameter_queue_url = ParameterString(name=\"parameter_queue_url\", default_value=queue_url)\n",
    "parameter_eoj_input_config = ParameterString(name=\"parameter_eoj_input_config\", default_value=\"\")\n",
    "parameter_cr_eoj_config = ParameterString(name=\"parameter_cr_eoj_config\", default_value=\"\")\n",
    "parameter_s_eoj_config = ParameterString(name=\"parameter_s_eoj_config\", default_value=\"\")\n",
    "parameter_eoj_output_config = ParameterString(name=\"parameter_eoj_output_config\", default_value=\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "With all the elements in place, we can now start creating our steps with SageMaker Pipelines..."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from sagemaker.workflow.lambda_step import LambdaStep, LambdaOutput, LambdaOutputTypeEnum\n",
    "from sagemaker.lambda_helper import Lambda\n",
    "\n",
    "step_lambda_cr = LambdaStep(\n",
    "    name=\"CloudRemovalStep\",\n",
    "    lambda_func=Lambda(function_arn=function_arn),\n",
    "    inputs={\n",
    "        \"role\": parameter_role,\n",
    "        \"region\": parameter_region,\n",
    "        \"eoj_input_config\": parameter_eoj_input_config,\n",
    "        \"eoj_config\": parameter_cr_eoj_config,\n",
    "        \"eoj_name\": f'cloudremoval-{datetime.now().strftime(\"%Y-%m-%d-%H-%M\")}',\n",
    "    },\n",
    "    outputs=[\n",
    "        LambdaOutput(output_name=\"statusCode\", output_type=LambdaOutputTypeEnum.String),\n",
    "        LambdaOutput(output_name=\"eoj_arn\", output_type=LambdaOutputTypeEnum.String),\n",
    "    ],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from sagemaker.workflow.callback_step import CallbackStep, CallbackOutput, CallbackOutputTypeEnum\n",
    "\n",
    "step_callback_cr = CallbackStep(\n",
    "    name=\"CloudRemovalCallbackStep\",\n",
    "    depends_on=[\"CloudRemovalStep\"],\n",
    "    sqs_queue_url=parameter_queue_url,\n",
    "    inputs={\n",
    "        \"role\": parameter_role,\n",
    "        \"region\": parameter_region,\n",
    "        \"eoj_arn\": step_lambda_cr.properties.Outputs[\"eoj_arn\"],\n",
    "    },\n",
    "    outputs=[\n",
    "        CallbackOutput(output_name=\"eoj_status\", output_type=CallbackOutputTypeEnum.String),\n",
    "    ],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "step_lambda_s = LambdaStep(\n",
    "    name=\"StackingStep\",\n",
    "    depends_on=[\"CloudRemovalCallbackStep\"],\n",
    "    lambda_func=Lambda(function_arn=function_arn),\n",
    "    inputs={\n",
    "        \"role\": parameter_role,\n",
    "        \"region\": parameter_region,\n",
    "        \"eoj_input_config\": step_lambda_cr.properties.Outputs[\"eoj_arn\"],\n",
    "        \"eoj_config\": parameter_s_eoj_config,\n",
    "        \"eoj_name\": f'stacking-{datetime.now().strftime(\"%Y-%m-%d-%H-%M\")}',\n",
    "    },\n",
    "    outputs=[\n",
    "        LambdaOutput(output_name=\"statusCode\", output_type=LambdaOutputTypeEnum.String),\n",
    "        LambdaOutput(output_name=\"eoj_arn\", output_type=LambdaOutputTypeEnum.String),\n",
    "    ],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "step_callback_s = CallbackStep(\n",
    "    name=\"StackingCallbackStep\",\n",
    "    depends_on=[\"StackingStep\"],\n",
    "    sqs_queue_url=parameter_queue_url,\n",
    "    inputs={\n",
    "        \"role\": parameter_role,\n",
    "        \"region\": parameter_region,\n",
    "        \"eoj_arn\": step_lambda_s.properties.Outputs[\"eoj_arn\"],\n",
    "    },\n",
    "    outputs=[\n",
    "        CallbackOutput(output_name=\"statusJob\", output_type=CallbackOutputTypeEnum.String),\n",
    "    ],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "step_lambda_ex = LambdaStep(\n",
    "    name=\"ExportStep\",\n",
    "    depends_on=[\"StackingCallbackStep\"],\n",
    "    lambda_func=Lambda(function_arn=function_arn),\n",
    "    inputs={\n",
    "        \"eoj_arn\": step_lambda_s.properties.Outputs[\"eoj_arn\"],\n",
    "        \"role\": parameter_role,\n",
    "        \"region\": parameter_region,\n",
    "        \"eoj_output_config\": parameter_eoj_output_config,\n",
    "    },\n",
    "    outputs=[\n",
    "        LambdaOutput(output_name=\"statusCode\", output_type=LambdaOutputTypeEnum.String),\n",
    "        LambdaOutput(output_name=\"eoj_arn\", output_type=LambdaOutputTypeEnum.String),\n",
    "    ],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "step_callback_ex = CallbackStep(\n",
    "    name=\"ExportCallbackStep\",\n",
    "    depends_on=[\"ExportStep\"],\n",
    "    sqs_queue_url=parameter_queue_url,\n",
    "    inputs={\n",
    "        \"role\": parameter_role,\n",
    "        \"region\": parameter_region,\n",
    "        \"eoj_arn\": step_lambda_ex.properties.Outputs[\"eoj_arn\"],\n",
    "    },\n",
    "    outputs=[\n",
    "        CallbackOutput(output_name=\"statusJob\", output_type=CallbackOutputTypeEnum.String),\n",
    "    ],\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, we can define our pipeline based on the steps and parameters created before."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from sagemaker.workflow.pipeline import Pipeline\n",
    "\n",
    "pipeline_name = \"GeospatialPipeline\"\n",
    "\n",
    "pipeline = Pipeline(\n",
    "    name=pipeline_name,\n",
    "    parameters=[\n",
    "        parameter_role,\n",
    "        parameter_region,\n",
    "        parameter_queue_url,\n",
    "        parameter_eoj_input_config,\n",
    "        parameter_cr_eoj_config,\n",
    "        parameter_s_eoj_config,\n",
    "        parameter_eoj_output_config,\n",
    "    ],\n",
    "    steps=[\n",
    "        step_lambda_cr,\n",
    "        step_callback_cr,\n",
    "        step_lambda_s,\n",
    "        step_callback_s,\n",
    "        step_lambda_ex,\n",
    "        step_callback_ex,\n",
    "    ],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true,
    "tags": []
   },
   "outputs": [],
   "source": [
    "definition = json.loads(pipeline.definition())\n",
    "definition"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "pipeline.upsert(role_arn=role)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let us test our pipeline by defining some values for our parameters and running an execution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Replace with the data collection of interest...\n",
    "data_collection_arn = \"arn:aws:sagemaker-geospatial:us-west-2:378778860802:raster-data-collection/public/nmqj48dcu3g7ayw8\"\n",
    "\n",
    "# Replace with the coordinates of interest...\n",
    "coordinates = [\n",
    "    [9.181602157004177, 53.14038825707946],\n",
    "    [9.181602157004177, 52.30629767547948],\n",
    "    [10.587520893823973, 52.30629767547948],\n",
    "    [10.587520893823973, 53.14038825707946],\n",
    "    [9.181602157004177, 53.14038825707946],\n",
    "]\n",
    "# Replace with the time-range of interest...\n",
    "time_start = \"2022-03-05T12:00:00Z\"\n",
    "time_end = \"2022-03-15T12:00:00Z\"\n",
    "\n",
    "eoj_input_config = {\n",
    "    \"RasterDataCollectionQuery\": {\n",
    "        \"RasterDataCollectionArn\": data_collection_arn,\n",
    "        \"AreaOfInterest\": {\n",
    "            \"AreaOfInterestGeometry\": {\"PolygonGeometry\": {\"Coordinates\": [coordinates]}}\n",
    "        },\n",
    "        \"TimeRangeFilter\": {\"StartTime\": time_start, \"EndTime\": time_end},\n",
    "        \"PropertyFilters\": {\n",
    "            \"Properties\": [{\"Property\": {\"EoCloudCover\": {\"LowerBound\": 0, \"UpperBound\": 2}}}]\n",
    "        },\n",
    "    }\n",
    "}\n",
    "\n",
    "cr_eoj_config = {\n",
    "    \"CloudRemovalConfig\": {\"AlgorithmName\": \"INTERPOLATION\", \"InterpolationValue\": \"-9999\"}\n",
    "}\n",
    "\n",
    "s_eoj_config = {\n",
    "    \"StackConfig\": {\n",
    "        \"OutputResolution\": {\"Predefined\": \"HIGHEST\"},\n",
    "        \"TargetBands\": [\"red\", \"green\", \"blue\"],\n",
    "    }\n",
    "}\n",
    "\n",
    "eoj_output_config = {\"S3Data\": {\"S3Uri\": f\"s3://{bucket}/{prefix}/export/\", \"KmsKeyId\": \"\"}}\n",
    "\n",
    "execution = pipeline.start(\n",
    "    parameters=dict(\n",
    "        parameter_role=role,\n",
    "        parameter_region=region,\n",
    "        parameter_queue_url=queue_url,\n",
    "        parameter_eoj_input_config=eoj_input_config,\n",
    "        parameter_cr_eoj_config=cr_eoj_config,\n",
    "        parameter_s_eoj_config=s_eoj_config,\n",
    "        parameter_eoj_output_config=eoj_output_config,\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At this point, you can go to the SageMaker Resources tab in the left menu in Studio and check the Pipelines.\n",
    "\n",
    "You should be able to see our \"GeospatialPipeline\" in the list and double-cling on it for checking the details of the execution."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "-------\n",
    "\n",
    "### Clean-up"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once done, uncomment and run the following cells for deleting any resources that could incur in costs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Delete the SQS queue\n",
    "# sqs_client = boto3.client(\"sqs\", region_name=region)\n",
    "# sqs_client.delete_queue(\n",
    "#    QueueUrl=sqs_client.get_queue_url(QueueName=f\"geospatial-queue\")[\"QueueUrl\"]\n",
    "# )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Delete Lambda function\n",
    "# lambda_client = boto3.client(\"lambda\", region_name=region)\n",
    "# lambda_client.delete_function(FunctionName=\"geospatial-lambda\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Delete the SageMaker Pipeline\n",
    "# sagemaker_client = boto3.client(\"sagemaker\", region_name=region)\n",
    "# sagemaker_client.delete_pipeline(PipelineName=\"GeospatialPipeline\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Notebook CI Test Results\n",
    "\n",
    "This notebook was tested in multiple regions. The test results are as follows, except for us-west-2 which is shown at the top of the notebook.\n",
    "\n",
    "![This us-east-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/us-east-1/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This us-east-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/us-east-2/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This us-west-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/us-west-1/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This ca-central-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/ca-central-1/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This sa-east-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/sa-east-1/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This eu-west-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/eu-west-1/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This eu-west-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/eu-west-2/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This eu-west-3 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/eu-west-3/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This eu-central-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/eu-central-1/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This eu-north-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/eu-north-1/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This ap-southeast-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/ap-southeast-1/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This ap-southeast-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/ap-southeast-2/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This ap-northeast-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/ap-northeast-1/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This ap-northeast-2 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/ap-northeast-2/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n",
    "\n",
    "![This ap-south-1 badge failed to load. Check your device's internet connectivity, otherwise the service is currently unavailable](https://h75twx4l60.execute-api.us-west-2.amazonaws.com/sagemaker-nb/ap-south-1/sagemaker-geospatial|digital-farming-pipelines|pipelines-sagemaker-geospatial.ipynb)\n"
   ]
  }
 ],
 "metadata": {
  "instance_type": "ml.t3.medium",
  "kernelspec": {
   "display_name": "Python 3 (Data Science 3.0)",
   "language": "python",
   "name": "python3__SAGEMAKER_INTERNAL__arn:aws:sagemaker:us-west-2:236514542706:image/sagemaker-data-science-310-v1"
  },
  "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.10.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}