Subchapter 37.100

Framework Processing

Running Release Tests

Scanning With AWS Security Agent

Setup DevOps Agent

Setup Security Agent

Setup

Threat Modeling With AWS Security Agent

LLM To Bedrock

Operate On AWS

Prompt Library For Startups

Start Building For Startups

Tf Best Practices

Code Output Guidereferences

Dataset Transformation Code

Sagemaker Dataset Formats

Transformation Tools

Overviewdirectory-management

Overviewendpoint-diagnostics

104 more in the appendix

Framework Processing

Auth

Billing And Cost Management

Messaging And Streaming

Well Architected Review

Launch With AWS

Signing In To AWS

Creating Amazon Aurora Db Cluster With Instances

Shieldadvanced

Resilience Program Design

Connecting Lambda To DynamoDB

Creating API Gateway Stage

Debugging Lambda Timeouts

Deploying Custom Domain REST API

Processing S3 Uploads With Step Functions

Raw file

aws/agent-toolkit-for-aws

references/processing-jobs/code_templates/framework_processing.py

Python101 lines3 KB

1# Cell 0 [markdown]: Framework Processing
2
3# Cell 1: Install dependencies
4%pip install --upgrade 'sagemaker>=3.22.0,<4.0' boto3 -q  # NOTEBOOK_ONLY
5
6# Cell 2: Configure the processing job
7import boto3
8from sagemaker.core import Attribution, image_uris, set_attribution
9from sagemaker.core.helper.session_helper import Session, get_execution_role
10from sagemaker.core.processing import

framework_processing.py on GitHub

FrameworkProcessor, logs_for_processing_job

from sagemaker.core.shapes import (

ProcessingInput,

ProcessingOutput,

ProcessingS3Input,

ProcessingS3Output,

)

set_attribution(Attribution.SAGEMAKER_AGENT_PLUGIN)

REGION = ""

ROLE_ARN = "[ROLE_ARN]"

FRAMEWORK = "[FRAMEWORK]"

FRAMEWORK_VERSION = ""

PYTHON_VERSION = ""

SOURCE_DIR = "[SOURCE_DIR]"

ENTRY_SCRIPT = "[ENTRY_SCRIPT]"

REQUIREMENTS = "[REQUIREMENTS]"

CUSTOM_IMAGE_URI = ""

INPUT_S3_URI = "[INPUT_S3_URI]"

OUTPUT_S3_URI = "[OUTPUT_S3_URI]"

INSTANCE_TYPE = "[INSTANCE_TYPE]"

INSTANCE_COUNT = 1

MAX_RUNTIME_SECONDS = 86400

BASE_JOB_NAME = "[BASE_JOB_NAME]"

ARGUMENTS = []

WAIT_FOR_COMPLETION = False

boto_session = boto3.Session(region_name=REGION or None)

region = boto_session.region_name

sm_client = boto_session.client("sagemaker")

sm_session = Session(boto_session=boto_session, sagemaker_client=sm_client)

role = ROLE_ARN or get_execution_role()

image_uri = CUSTOM_IMAGE_URI or image_uris.retrieve(

framework=FRAMEWORK,

region=region,

version=FRAMEWORK_VERSION or None,

py_version=PYTHON_VERSION or None,

instance_type=INSTANCE_TYPE,

)

print(f"Resolved processing image URI: {image_uri}")

# Cell 3: Build and submit the processor

processor = FrameworkProcessor(

image_uri=image_uri,

role=role,

command=["python3"],

instance_type=INSTANCE_TYPE,

instance_count=INSTANCE_COUNT,

max_runtime_in_seconds=MAX_RUNTIME_SECONDS,

base_job_name=BASE_JOB_NAME,

sagemaker_session=sm_session,

)

processor.run(

code=ENTRY_SCRIPT,

source_dir=SOURCE_DIR,

requirements=REQUIREMENTS or None,

inputs=[

ProcessingInput(

input_name="input",

s3_input=ProcessingS3Input(

s3_uri=INPUT_S3_URI,

local_path="/opt/ml/processing/input",

s3_data_type="S3Prefix",

s3_input_mode="File",

),

)

],

outputs=[

ProcessingOutput(

output_name="output",

s3_output=ProcessingS3Output(

s3_uri=OUTPUT_S3_URI,

local_path="/opt/ml/processing/output",

s3_upload_mode="EndOfJob",

),

)

],

arguments=ARGUMENTS,

wait=False,

logs=False,

)

# Cell 4: Report status and optionally wait

job_name = processor.latest_job.processing_job_name

job_arn = sm_client.describe_processing_job(ProcessingJobName=job_name)["ProcessingJobArn"]