AWSCDK::StepFunctionsTasks

293 types

Tasks for AWS Step Functions

AWS Step Functions is a web service that enables you to coordinate the components of distributed applications and microservices using visual workflows. You build applications from individual components that each perform a discrete function, or task, allowing you to scale and change applications quickly.

A Task state represents a single unit of work performed by a state machine. All work in your state machine is performed by tasks. This module contains a collection of classes that allow you to call various AWS services from your Step Functions state machine.

Be sure to familiarize yourself with the aws-stepfunctions module documentation first.

This module is part of the AWS Cloud Development Kit project.

Tasks for AWS Step Functions

Paths

Learn more about input and output processing in Step Functions here

Evaluate Expression

Use the EvaluateExpression to perform simple operations referencing state paths. The expression referenced in the task will be evaluated in a Lambda function (eval()). This allows you to not have to write Lambda code for simple operations.

Example: convert a wait time from milliseconds to seconds, concat this in a message and wait:

convert_to_seconds = AWSCDK::StepFunctionsTasks::EvaluateExpression.new(self, "Convert to seconds", {
    expression: "$.waitMilliseconds / 1000",
    result_path: "$.waitSeconds",
})

create_message = AWSCDK::StepFunctionsTasks::EvaluateExpression.new(self, "Create message", {
    # Note: this is a string inside a string.
    expression: "`Now waiting ${$.waitSeconds} seconds...`",
    runtime: AWSCDK::Lambda::Runtime.NODEJS_LATEST,
    result_path: "$.message",
})

publish_message = AWSCDK::StepFunctionsTasks::SNSPublish.new(self, "Publish message", {
    topic: AWSCDK::SNS::Topic.new(self, "cool-topic"),
    message: AWSCDK::StepFunctions::TaskInput.from_json_path_at("$.message"),
    result_path: "$.sns",
})

wait = AWSCDK::StepFunctions::Wait.new(self, "Wait", {
    time: AWSCDK::StepFunctions::WaitTime.seconds_path("$.waitSeconds"),
})

AWSCDK::StepFunctions::StateMachine.new(self, "StateMachine", {
    definition: convert_to_seconds._next(create_message)._next(publish_message)._next(wait),
})

The EvaluateExpression supports a runtime prop to specify the Lambda runtime to use to evaluate the expression. Currently, only runtimes of the Node.js family are supported.

The EvaluateExpression also supports an architecture prop to specify the Lambda architecture. This can be useful when migrating to ARM64 or when running integration tests on ARM64 systems.

convert_to_seconds_arm64 = AWSCDK::StepFunctionsTasks::EvaluateExpression.new(self, "Convert to seconds", {
    expression: "$.waitMilliseconds / 1000",
    architecture: AWSCDK::Lambda::Architecture.ARM_64,
})

API Gateway

Step Functions supports API Gateway through the service integration pattern.

HTTP APIs are designed for low-latency, cost-effective integrations with AWS services, including AWS Lambda, and HTTP endpoints. HTTP APIs support OIDC and OAuth 2.0 authorization, and come with built-in support for CORS and automatic deployments. Previous-generation REST APIs currently offer more features. More details can be found here.

Call REST API Endpoint

The CallApiGatewayRestApiEndpoint calls the REST API endpoint.

require 'aws-cdk-lib'

rest_api = AWSCDK::APIGateway::RestAPI.new(self, "MyRestApi")

invoke_task = AWSCDK::StepFunctionsTasks::CallAPIGatewayRestAPIEndpoint.new(self, "Call REST API", {
    api: rest_api,
    stage_name: "prod",
    method: AWSCDK::StepFunctionsTasks::HttpMethod::GET,
})

By default, the API endpoint URI will be constructed using the AWS region of the stack in which the provided api is created.

To construct the endpoint with a different region, use the region parameter:

require 'aws-cdk-lib'

rest_api = AWSCDK::APIGateway::RestAPI.new(self, "MyRestApi")
invoke_task = AWSCDK::StepFunctionsTasks::CallAPIGatewayRestAPIEndpoint.new(self, "Call REST API", {
    api: rest_api,
    stage_name: "prod",
    method: AWSCDK::StepFunctionsTasks::HttpMethod::GET,
    region: "us-west-2",
})

Be aware that the header values must be arrays. When passing the Task Token in the headers field WAIT_FOR_TASK_TOKEN integration, use JsonPath.array() to wrap the token in an array:

require 'aws-cdk-lib'
api = nil # AWSCDK::APIGateway::RestAPI


AWSCDK::StepFunctionsTasks::CallAPIGatewayRestAPIEndpoint.new(self, "Endpoint", {
    api: api,
    stage_name: "Stage",
    method: AWSCDK::StepFunctionsTasks::HttpMethod::PUT,
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::WAIT_FOR_TASK_TOKEN,
    headers: AWSCDK::StepFunctions::TaskInput.from_object({
        TaskToken: AWSCDK::StepFunctions::JsonPath.array(AWSCDK::StepFunctions::JsonPath.task_token),
    }),
})

Call HTTP API Endpoint

The CallApiGatewayHttpApiEndpoint calls the HTTP API endpoint.

require 'aws-cdk-lib'

http_api = AWSCDK::APIGatewayv2::HttpAPI.new(self, "MyHttpApi")

invoke_task = AWSCDK::StepFunctionsTasks::CallAPIGatewayHttpAPIEndpoint.new(self, "Call HTTP API", {
    api_id: http_api.api_id,
    api_stack: AWSCDK::Stack.of(http_api),
    method: AWSCDK::StepFunctionsTasks::HttpMethod::GET,
})

AWS SDK

Step Functions supports calling AWS service's API actions through the service integration pattern.

You can use Step Functions' AWS SDK integrations to call any of the over two hundred AWS services directly from your state machine, giving you access to over nine thousand API actions.

my_bucket = nil # AWSCDK::S3::Bucket

get_object = AWSCDK::StepFunctionsTasks::CallAWSService.new(self, "GetObject", {
    service: "s3",
    action: "getObject",
    parameters: {
        Bucket: my_bucket.bucket_name,
        Key: AWSCDK::StepFunctions::JsonPath.string_at("$.key"),
    },
    iam_resources: [my_bucket.arn_for_objects("*")],
})

Use camelCase for actions and PascalCase for parameter names.

The task automatically adds an IAM statement to the state machine role's policy based on the service and action called. The resources for this statement must be specified in iam_resources.

Use the iam_action prop to manually specify the IAM action name in the case where the IAM action name does not match with the API service/action name:

list_buckets = AWSCDK::StepFunctionsTasks::CallAWSService.new(self, "ListBuckets", {
    service: "s3",
    action: "listBuckets",
    iam_resources: ["*"],
    iam_action: "s3:ListAllMyBuckets",
})

Use the additional_iam_statements prop to pass additional IAM statements that will be added to the state machine role's policy. Use it in the case where the call requires more than a single statement to be executed:

detect_labels = AWSCDK::StepFunctionsTasks::CallAWSService.new(self, "DetectLabels", {
    service: "rekognition",
    action: "detectLabels",
    iam_resources: ["*"],
    additional_iam_statements: [
        AWSCDK::IAM::PolicyStatement.new({
            actions: ["s3:getObject"],
            resources: ["arn:aws:s3:::amzn-s3-demo-bucket/*"],
        }),
    ],
})

Cross-region AWS API call

You can call AWS API in a different region from your state machine by using the CallAwsServiceCrossRegion construct. In addition to the properties for CallAwsService construct, you have to set region property to call the API.

my_bucket = nil # AWSCDK::S3::Bucket

get_object = AWSCDK::StepFunctionsTasks::CallAWSServiceCrossRegion.new(self, "GetObject", {
    region: "ap-northeast-1",
    service: "s3",
    action: "getObject",
    parameters: {
        Bucket: my_bucket.bucket_name,
        Key: AWSCDK::StepFunctions::JsonPath.string_at("$.key"),
    },
    iam_resources: [my_bucket.arn_for_objects("*")],
})

Other properties such as additional_iam_statements can be used in the same way as the CallAwsService task.

Note that when you use integrationPattern.WAIT_FOR_TASK_TOKEN, the output path changes under Payload property.

Athena

Step Functions supports Athena through the service integration pattern.

StartQueryExecution

The StartQueryExecution API runs the SQL query statement.

start_query_execution_job = AWSCDK::StepFunctionsTasks::AthenaStartQueryExecution.new(self, "Start Athena Query", {
    query_string: AWSCDK::StepFunctions::JsonPath.string_at("$.queryString"),
    query_execution_context: {
        database_name: "mydatabase",
    },
    result_configuration: {
        encryption_configuration: {
            encryption_option: AWSCDK::StepFunctionsTasks::EncryptionOption::S3_MANAGED,
        },
        output_location: {
            bucket_name: "amzn-s3-demo-bucket",
            object_key: "folder",
        },
    },
    execution_parameters: ["param1", "param2"],
})

You can reuse the query results by setting the result_reuse_configuration_max_age property.

start_query_execution_job = AWSCDK::StepFunctionsTasks::AthenaStartQueryExecution.new(self, "Start Athena Query", {
    query_string: AWSCDK::StepFunctions::JsonPath.string_at("$.queryString"),
    query_execution_context: {
        database_name: "mydatabase",
    },
    result_configuration: {
        encryption_configuration: {
            encryption_option: AWSCDK::StepFunctionsTasks::EncryptionOption::S3_MANAGED,
        },
        output_location: {
            bucket_name: "query-results-bucket",
            object_key: "folder",
        },
    },
    execution_parameters: ["param1", "param2"],
    result_reuse_configuration_max_age: AWSCDK::Duration.minutes(100),
})

GetQueryExecution

The GetQueryExecution API gets information about a single execution of a query.

get_query_execution_job = AWSCDK::StepFunctionsTasks::AthenaGetQueryExecution.new(self, "Get Query Execution", {
    query_execution_id: AWSCDK::StepFunctions::JsonPath.string_at("$.QueryExecutionId"),
})

GetQueryResults

The GetQueryResults API that streams the results of a single query execution specified by QueryExecutionId from S3.

get_query_results_job = AWSCDK::StepFunctionsTasks::AthenaGetQueryResults.new(self, "Get Query Results", {
    query_execution_id: AWSCDK::StepFunctions::JsonPath.string_at("$.QueryExecutionId"),
})

StopQueryExecution

The StopQueryExecution API that stops a query execution.

stop_query_execution_job = AWSCDK::StepFunctionsTasks::AthenaStopQueryExecution.new(self, "Stop Query Execution", {
    query_execution_id: AWSCDK::StepFunctions::JsonPath.string_at("$.QueryExecutionId"),
})

Batch

Step Functions supports Batch through the service integration pattern.

SubmitJob

The SubmitJob API submits an AWS Batch job from a job definition.

require 'aws-cdk-lib'
batch_job_definition = nil # AWSCDK::Batch::ECSJobDefinition
batch_queue = nil # AWSCDK::Batch::JobQueue


task = AWSCDK::StepFunctionsTasks::BatchSubmitJob.new(self, "Submit Job", {
    job_definition_arn: batch_job_definition.job_definition_arn,
    job_name: "MyJob",
    job_queue_arn: batch_queue.job_queue_arn,
})

Bedrock

Step Functions supports Bedrock through the service integration pattern.

InvokeModel

The InvokeModel API invokes the specified Bedrock model to run inference using the input provided. The format of the input body and the response body depend on the model selected.

require 'aws-cdk-lib'


model = AWSCDK::Bedrock::FoundationModel.from_foundation_model_id(self, "Model", AWSCDK::Bedrock::FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1)

task = AWSCDK::StepFunctionsTasks::BedrockInvokeModel.new(self, "Prompt Model", {
    model: model,
    body: AWSCDK::StepFunctions::TaskInput.from_object({
        input_text: "Generate a list of five first names.",
        text_generation_config: {
            max_token_count: 100,
            temperature: 1,
        },
    }),
    result_selector: {
        names: AWSCDK::StepFunctions::JsonPath.string_at("$.Body.results[0].outputText"),
    },
})

Using Input Path for S3 URI

Provide S3 URI as an input or output path to invoke a model

To specify the S3 URI as JSON path to your input or output fields, use props s3_input_uri and s3_output_uri under BedrockInvokeModelProps and set feature flag @aws-cdk/aws-stepfunctions-tasks:useNewS3UriParametersForBedrockInvokeModelTask to true.

If this flag is not enabled, the code will populate the S3Uri using InputPath and OutputPath fields, which is not recommended.

require 'aws-cdk-lib'


model = AWSCDK::Bedrock::FoundationModel.from_foundation_model_id(self, "Model", AWSCDK::Bedrock::FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1)

task = AWSCDK::StepFunctionsTasks::BedrockInvokeModel.new(self, "Prompt Model", {
    model: model,
    input: {s3_input_uri: AWSCDK::StepFunctions::JsonPath.string_at("$.prompt")},
    output: {s3_output_uri: AWSCDK::StepFunctions::JsonPath.string_at("$.prompt")},
})

Using Input Path

Provide S3 URI as an input or output path to invoke a model

Currently, input and output Path provided in the BedrockInvokeModelProps input is defined as S3URI field under task definition of state machine. To modify the existing behaviour, set @aws-cdk/aws-stepfunctions-tasks:useNewS3UriParametersForBedrockInvokeModelTask to true.

If this feature flag is enabled, S3URI fields will be generated from other Props(s3_input_uri and s3_output_uri), and the given inputPath, OutputPath will be rendered as it is in the JSON task definition.

If the feature flag is set to false, the behavior will be to populate the S3Uri using the InputPath and OutputPath fields, which is not recommended.

require 'aws-cdk-lib'


model = AWSCDK::Bedrock::FoundationModel.from_foundation_model_id(self, "Model", AWSCDK::Bedrock::FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1)

task = AWSCDK::StepFunctionsTasks::BedrockInvokeModel.new(self, "Prompt Model", {
    model: model,
    input_path: AWSCDK::StepFunctions::JsonPath.string_at("$.prompt"),
    output_path: AWSCDK::StepFunctions::JsonPath.string_at("$.prompt"),
})

You can apply a guardrail to the invocation by setting guardrail.

require 'aws-cdk-lib'


model = AWSCDK::Bedrock::FoundationModel.from_foundation_model_id(self, "Model", AWSCDK::Bedrock::FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1)

task = AWSCDK::StepFunctionsTasks::BedrockInvokeModel.new(self, "Prompt Model with guardrail", {
    model: model,
    body: AWSCDK::StepFunctions::TaskInput.from_object({
        input_text: "Generate a list of five first names.",
        text_generation_config: {
            max_token_count: 100,
            temperature: 1,
        },
    }),
    guardrail: AWSCDK::StepFunctionsTasks::Guardrail.enable("guardrailId", 1),
    result_selector: {
        names: AWSCDK::StepFunctions::JsonPath.string_at("$.Body.results[0].outputText"),
    },
})

createModelCustomizationJob

The CreateModelCustomizationJob API creates a fine-tuning job to customize a base model.

require 'aws-cdk-lib'

output_bucket = nil # AWSCDK::S3::IBucket
training_bucket = nil # AWSCDK::S3::IBucket
validation_bucket = nil # AWSCDK::S3::IBucket
kms_key = nil # AWSCDK::KMS::IKey
vpc = nil # AWSCDK::EC2::IVPC


model = AWSCDK::Bedrock::FoundationModel.from_foundation_model_id(self, "Model", AWSCDK::Bedrock::FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1)

task = AWSCDK::StepFunctionsTasks::BedrockCreateModelCustomizationJob.new(self, "CreateModelCustomizationJob", {
    base_model: model,
    client_request_token: "MyToken",
    customization_type: AWSCDK::StepFunctionsTasks::CustomizationType::FINE_TUNING,
    custom_model_kms_key: kms_key,
    custom_model_name: "MyCustomModel",
     # required
    custom_model_tags: [{key: "key1", value: "value1"}],
    hyper_parameters: {
        batch_size: "10",
    },
    job_name: "MyCustomizationJob",
     # required
    job_tags: [{key: "key2", value: "value2"}],
    output_data: {
        bucket: output_bucket,
         # required
        path: "output-data/",
    },
    training_data: {
        bucket: training_bucket,
        path: "training-data/data.json",
    },
     # required
    # If you don't provide validation data, you have to specify `Evaluation percentage` hyperparameter.
    validation_data: [
        {
            bucket: validation_bucket,
            path: "validation-data/data.json",
        },
    ],
    vpc_config: {
        security_groups: [AWSCDK::EC2::SecurityGroup.new(self, "SecurityGroup", {vpc: vpc})],
        subnets: vpc.private_subnets,
    },
})

CodeBuild

Step Functions supports CodeBuild through the service integration pattern.

StartBuild

StartBuild starts a CodeBuild Project by Project Name.

require 'aws-cdk-lib'


codebuild_project = AWSCDK::CodeBuild::Project.new(self, "Project", {
    project_name: "MyTestProject",
    build_spec: AWSCDK::CodeBuild::BuildSpec.from_object({
        version: "0.2",
        phases: {
            build: {
                commands: [
                    "echo \"Hello, CodeBuild!\"",
                ],
            },
        },
    }),
})

task = AWSCDK::StepFunctionsTasks::CodeBuildStartBuild.new(self, "Task", {
    project: codebuild_project,
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::RUN_JOB,
    environment_variables_override: {
        ZONE: {
            type: AWSCDK::CodeBuild::BuildEnvironmentVariableType::PLAINTEXT,
            value: AWSCDK::StepFunctions::JsonPath.string_at("$.envVariables.zone"),
        },
    },
})

StartBuildBatch

StartBuildBatch starts a batch CodeBuild for a project by project name. It is necessary to enable the batch build feature in the CodeBuild project.

require 'aws-cdk-lib'


project = AWSCDK::CodeBuild::Project.new(self, "Project", {
    project_name: "MyTestProject",
    build_spec: AWSCDK::CodeBuild::BuildSpec.from_object_to_yaml({
        version: 0.2,
        batch: {
            "build-list" => [
                {
                    identifier: "id",
                    buildspec: <<-'HERE'
                    version: 0.2
                    phases:
                      build:
                        commands:
                          - echo "Hello, from small!"
                    HERE,
                },
            ],
        },
    }),
})
project.enable_batch_builds

task = AWSCDK::StepFunctionsTasks::CodeBuildStartBuildBatch.new(self, "buildBatchTask", {
    project: project,
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::REQUEST_RESPONSE,
    environment_variables_override: {
        test: {
            type: AWSCDK::CodeBuild::BuildEnvironmentVariableType::PLAINTEXT,
            value: "testValue",
        },
    },
})

Note: enable_batch_builds() will do nothing for imported projects. If you are using an imported project, you must ensure that the project is already configured for batch builds.

DynamoDB

You can call DynamoDB APIs from a Task state. Read more about calling DynamoDB APIs here

GetItem

The GetItem operation returns a set of attributes for the item with the given primary key.

my_table = nil # AWSCDK::DynamoDB::Table

AWSCDK::StepFunctionsTasks::DynamoGetItem.new(self, "Get Item", {
    key: {message_id: AWSCDK::StepFunctionsTasks::DynamoAttributeValue.from_string("message-007")},
    table: my_table,
})

PutItem

The PutItem operation creates a new item, or replaces an old item with a new item.

my_table = nil # AWSCDK::DynamoDB::Table

AWSCDK::StepFunctionsTasks::DynamoPutItem.new(self, "PutItem", {
    item: {
        MessageId: AWSCDK::StepFunctionsTasks::DynamoAttributeValue.from_string("message-007"),
        Text: AWSCDK::StepFunctionsTasks::DynamoAttributeValue.from_string(AWSCDK::StepFunctions::JsonPath.string_at("$.bar")),
        TotalCount: AWSCDK::StepFunctionsTasks::DynamoAttributeValue.from_number(10),
    },
    table: my_table,
})

DeleteItem

The DeleteItem operation deletes a single item in a table by primary key.

my_table = nil # AWSCDK::DynamoDB::Table

AWSCDK::StepFunctionsTasks::DynamoDeleteItem.new(self, "DeleteItem", {
    key: {MessageId: AWSCDK::StepFunctionsTasks::DynamoAttributeValue.from_string("message-007")},
    table: my_table,
    result_path: AWSCDK::StepFunctions::JsonPath.DISCARD,
})

UpdateItem

The UpdateItem operation edits an existing item's attributes, or adds a new item to the table if it does not already exist.

my_table = nil # AWSCDK::DynamoDB::Table

AWSCDK::StepFunctionsTasks::DynamoUpdateItem.new(self, "UpdateItem", {
    key: {
        MessageId: AWSCDK::StepFunctionsTasks::DynamoAttributeValue.from_string("message-007"),
    },
    table: my_table,
    expression_attribute_values: {
        ":val" => AWSCDK::StepFunctionsTasks::DynamoAttributeValue.number_from_string(AWSCDK::StepFunctions::JsonPath.string_at("$.Item.TotalCount.N")),
        ":rand" => AWSCDK::StepFunctionsTasks::DynamoAttributeValue.from_number(20),
    },
    update_expression: "SET TotalCount = :val + :rand",
})

ECS

Step Functions supports ECS/Fargate through the service integration pattern.

RunTask

RunTask starts a new task using the specified task definition.

EC2

The EC2 launch type allows you to run your containerized applications on a cluster of Amazon EC2 instances that you manage.

When a task that uses the EC2 launch type is launched, Amazon ECS must determine where to place the task based on the requirements specified in the task definition, such as CPU and memory. Similarly, when you scale down the task count, Amazon ECS must determine which tasks to terminate. You can apply task placement strategies and constraints to customize how Amazon ECS places and terminates tasks. Learn more about task placement

The latest ACTIVE revision of the passed task definition is used for running the task.

The following example runs a job from a task definition on EC2

vpc = AWSCDK::EC2::VPC.from_lookup(self, "Vpc", {
    is_default: true,
})

cluster = AWSCDK::ECS::Cluster.new(self, "Ec2Cluster", {vpc: vpc})
cluster.add_capacity("DefaultAutoScalingGroup", {
    instance_type: AWSCDK::EC2::InstanceType.new("t2.micro"),
    vpc_subnets: {subnet_type: AWSCDK::EC2::SubnetType::PUBLIC},
})

task_definition = AWSCDK::ECS::TaskDefinition.new(self, "TD", {
    compatibility: AWSCDK::ECS::Compatibility::EC2,
})

task_definition.add_container("TheContainer", {
    image: AWSCDK::ECS::ContainerImage.from_registry("foo/bar"),
    memory_limit_mi_b: 256,
})

run_task = AWSCDK::StepFunctionsTasks::ECSRunTask.new(self, "Run", {
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::RUN_JOB,
    cluster: cluster,
    task_definition: task_definition,
    launch_target: AWSCDK::StepFunctionsTasks::ECSEC2LaunchTarget.new({
        placement_strategies: [
            AWSCDK::ECS::PlacementStrategy.spread_across_instances,
            AWSCDK::ECS::PlacementStrategy.packed_by_cpu,
            AWSCDK::ECS::PlacementStrategy.randomly,
        ],
        placement_constraints: [
            AWSCDK::ECS::PlacementConstraint.member_of("blieptuut"),
        ],
    }),
    propagated_tag_source: AWSCDK::ECS::PropagatedTagSource::TASK_DEFINITION,
})

Fargate

AWS Fargate is a serverless compute engine for containers that works with Amazon Elastic Container Service (ECS). Fargate makes it easy for you to focus on building your applications. Fargate removes the need to provision and manage servers, lets you specify and pay for resources per application, and improves security through application isolation by design. Learn more about Fargate

The Fargate launch type allows you to run your containerized applications without the need to provision and manage the backend infrastructure. Just register your task definition and Fargate launches the container for you. The latest ACTIVE revision of the passed task definition is used for running the task. Learn more about Fargate Versioning

The following example runs a job from a task definition on Fargate

vpc = AWSCDK::EC2::VPC.from_lookup(self, "Vpc", {
    is_default: true,
})

cluster = AWSCDK::ECS::Cluster.new(self, "FargateCluster", {vpc: vpc})

task_definition = AWSCDK::ECS::TaskDefinition.new(self, "TD", {
    memory_mi_b: "512",
    cpu: "256",
    compatibility: AWSCDK::ECS::Compatibility::FARGATE,
})

container_definition = task_definition.add_container("TheContainer", {
    image: AWSCDK::ECS::ContainerImage.from_registry("foo/bar"),
    memory_limit_mi_b: 256,
})

run_task = AWSCDK::StepFunctionsTasks::ECSRunTask.new(self, "RunFargate", {
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::RUN_JOB,
    cluster: cluster,
    task_definition: task_definition,
    assign_public_ip: true,
    container_overrides: [
        {
            container_definition: container_definition,
            environment: [{name: "SOME_KEY", value: AWSCDK::StepFunctions::JsonPath.string_at("$.SomeKey")}],
        },
    ],
    launch_target: AWSCDK::StepFunctionsTasks::ECSFargateLaunchTarget.new,
    propagated_tag_source: AWSCDK::ECS::PropagatedTagSource::TASK_DEFINITION,
})

Capacity Provider Options

The capacity_provider_options property allows you to configure the capacity provider strategy for both EC2 and Fargate launch targets.

When CapacityProviderOptions.custom() is used, you can specify a custom capacity provider strategy.
When CapacityProviderOptions.default() is used, the task uses the cluster's default capacity provider strategy.
If capacity_provider_options is not specified, the task uses the launch type (EC2 or FARGATE) without a capacity provider strategy.

vpc = AWSCDK::EC2::VPC.from_lookup(self, "Vpc", {
    is_default: true,
})

cluster = AWSCDK::ECS::Cluster.new(self, "FargateCluster", {vpc: vpc})

task_definition = AWSCDK::ECS::TaskDefinition.new(self, "TD", {
    memory_mi_b: "512",
    cpu: "256",
    compatibility: AWSCDK::ECS::Compatibility::FARGATE,
})

# Use custom() option - specify custom capacity provider strategy
run_task_with_custom = AWSCDK::StepFunctionsTasks::ECSRunTask.new(self, "RunTaskWithCustom", {
    cluster: cluster,
    task_definition: task_definition,
    launch_target: AWSCDK::StepFunctionsTasks::ECSFargateLaunchTarget.new({
        platform_version: AWSCDK::ECS::FargatePlatformVersion::VERSION1_4,
        capacity_provider_options: AWSCDK::StepFunctionsTasks::CapacityProviderOptions.custom([
            {capacity_provider: "FARGATE_SPOT", weight: 2, base: 1},
            {capacity_provider: "FARGATE", weight: 1},
        ]),
    }),
})

# Use default() option - uses cluster's default capacity provider strategy
run_task_with_default = AWSCDK::StepFunctionsTasks::ECSRunTask.new(self, "RunTaskWithDefault", {
    cluster: cluster,
    task_definition: task_definition,
    launch_target: AWSCDK::StepFunctionsTasks::ECSFargateLaunchTarget.new({
        platform_version: AWSCDK::ECS::FargatePlatformVersion::VERSION1_4,
        capacity_provider_options: AWSCDK::StepFunctionsTasks::CapacityProviderOptions.default,
    }),
})

Override CPU and Memory Parameter

By setting the property cpu or memoryMiB, you can override the Fargate or EC2 task instance size at runtime.

see: https://docs.aws.amazon.com/AmazonECS/latest/APIReference/API_TaskOverride.html

vpc = AWSCDK::EC2::VPC.from_lookup(self, "Vpc", {
    is_default: true,
})
cluster = AWSCDK::ECS::Cluster.new(self, "ECSCluster", {vpc: vpc})

task_definition = AWSCDK::ECS::TaskDefinition.new(self, "TD", {
    compatibility: AWSCDK::ECS::Compatibility::FARGATE,
    cpu: "256",
    memory_mi_b: "512",
})

task_definition.add_container("TheContainer", {
    image: AWSCDK::ECS::ContainerImage.from_registry("foo/bar"),
})

run_task = AWSCDK::StepFunctionsTasks::ECSRunTask.new(self, "Run", {
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::RUN_JOB,
    cluster: cluster,
    task_definition: task_definition,
    launch_target: AWSCDK::StepFunctionsTasks::ECSFargateLaunchTarget.new,
    cpu: "1024",
    memory_mi_b: "1048",
})

ECS enable Exec

By setting the property enable_execute_command to true, you can enable the ECS Exec feature for the task for either Fargate or EC2 launch types.

vpc = AWSCDK::EC2::VPC.from_lookup(self, "Vpc", {
    is_default: true,
})
cluster = AWSCDK::ECS::Cluster.new(self, "ECSCluster", {vpc: vpc})

task_definition = AWSCDK::ECS::TaskDefinition.new(self, "TD", {
    compatibility: AWSCDK::ECS::Compatibility::EC2,
})

task_definition.add_container("TheContainer", {
    image: AWSCDK::ECS::ContainerImage.from_registry("foo/bar"),
    memory_limit_mi_b: 256,
})

run_task = AWSCDK::StepFunctionsTasks::ECSRunTask.new(self, "Run", {
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::RUN_JOB,
    cluster: cluster,
    task_definition: task_definition,
    launch_target: AWSCDK::StepFunctionsTasks::ECSEC2LaunchTarget.new,
    enable_execute_command: true,
})

EMR

Step Functions supports Amazon EMR through the service integration pattern. The service integration APIs correspond to Amazon EMR APIs but differ in the parameters that are used.

Read more about the differences when using these service integrations.

Create Cluster

Creates and starts running a cluster (job flow). Corresponds to the run_job_flow API in EMR.

cluster_role = AWSCDK::IAM::Role.new(self, "ClusterRole", {
    assumed_by: AWSCDK::IAM::ServicePrincipal.new("ec2.amazonaws.com"),
})

service_role = AWSCDK::IAM::Role.new(self, "ServiceRole", {
    assumed_by: AWSCDK::IAM::ServicePrincipal.new("elasticmapreduce.amazonaws.com"),
})

auto_scaling_role = AWSCDK::IAM::Role.new(self, "AutoScalingRole", {
    assumed_by: AWSCDK::IAM::ServicePrincipal.new("elasticmapreduce.amazonaws.com"),
})

auto_scaling_role.assume_role_policy&.add_statements(
AWSCDK::IAM::PolicyStatement.new({
    effect: AWSCDK::IAM::Effect::ALLOW,
    principals: [
        AWSCDK::IAM::ServicePrincipal.new("application-autoscaling.amazonaws.com"),
    ],
    actions: [
        "sts:AssumeRole",
    ],
}))

AWSCDK::StepFunctionsTasks::EMRCreateCluster.new(self, "Create Cluster", {
    instances: {},
    cluster_role: cluster_role,
    name: AWSCDK::StepFunctions::TaskInput.from_json_path_at("$.ClusterName").value,
    service_role: service_role,
    auto_scaling_role: auto_scaling_role,
})

You can use the launch specification for On-Demand and Spot instances in the fleet.

AWSCDK::StepFunctionsTasks::EMRCreateCluster.new(self, "OnDemandSpecification", {
    instances: {
        instance_fleets: [
            {
                instance_fleet_type: AWSCDK::StepFunctionsTasks::EMRCreateCluster::InstanceRoleType::MASTER,
                launch_specifications: {
                    on_demand_specification: {
                        allocation_strategy: AWSCDK::StepFunctionsTasks::EMRCreateCluster::OnDemandAllocationStrategy::LOWEST_PRICE,
                    },
                },
            },
        ],
    },
    name: "OnDemandCluster",
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::RUN_JOB,
})

AWSCDK::StepFunctionsTasks::EMRCreateCluster.new(self, "SpotSpecification", {
    instances: {
        instance_fleets: [
            {
                instance_fleet_type: AWSCDK::StepFunctionsTasks::EMRCreateCluster::InstanceRoleType::MASTER,
                launch_specifications: {
                    spot_specification: {
                        allocation_strategy: AWSCDK::StepFunctionsTasks::EMRCreateCluster::SpotAllocationStrategy::CAPACITY_OPTIMIZED,
                        timeout_action: AWSCDK::StepFunctionsTasks::EMRCreateCluster::SpotTimeoutAction::TERMINATE_CLUSTER,
                        timeout: AWSCDK::Duration.minutes(5),
                    },
                },
            },
        ],
    },
    name: "SpotCluster",
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::RUN_JOB,
})

You can use the prioritized allocation strategy to specify instance type priorities for On-Demand instances:

AWSCDK::StepFunctionsTasks::EMRCreateCluster.new(self, "PrioritizedAllocation", {
    instances: {
        instance_fleets: [
            {
                instance_fleet_type: AWSCDK::StepFunctionsTasks::EMRCreateCluster::InstanceRoleType::CORE,
                instance_type_configs: [
                    {
                        instance_type: "m5.large",
                        priority: 0,
                    },
                    {
                        instance_type: "m5.xlarge",
                        priority: 1,
                    },
                ],
                launch_specifications: {
                    on_demand_specification: {
                        allocation_strategy: AWSCDK::StepFunctionsTasks::EMRCreateCluster::OnDemandAllocationStrategy::PRIORITIZED,
                    },
                },
                target_on_demand_capacity: 2,
            },
        ],
    },
    name: "PrioritizedCluster",
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::RUN_JOB,
})

You can customize EBS root device volume.

AWSCDK::StepFunctionsTasks::EMRCreateCluster.new(self, "Create Cluster", {
    instances: {},
    name: "ClusterName",
    ebs_root_volume_iops: 4000,
    ebs_root_volume_size: AWSCDK::Size.gibibytes(20),
    ebs_root_volume_throughput: 200,
})

If you want to run multiple steps in parallel, you can specify the step_concurrency_level property. The concurrency range is between 1 and 256 inclusive, where the default concurrency of 1 means no step concurrency is allowed. step_concurrency_level requires the EMR release label to be 5.28.0 or above.

AWSCDK::StepFunctionsTasks::EMRCreateCluster.new(self, "Create Cluster", {
    instances: {},
    name: AWSCDK::StepFunctions::TaskInput.from_json_path_at("$.ClusterName").value,
    step_concurrency_level: 10,
})

If you want to use an auto-termination policy, you can specify the auto_termination_policy_idle_timeout property. Specifies the amount of idle time after which the cluster automatically terminates. You can specify a minimum of 60 seconds and a maximum of 604800 seconds (seven days).

AWSCDK::StepFunctionsTasks::EMRCreateCluster.new(self, "Create Cluster", {
    instances: {},
    name: "ClusterName",
    auto_termination_policy_idle_timeout: AWSCDK::Duration.seconds(100),
})

If you want to use managed scaling, you can specify the managed_scaling_policy property.

AWSCDK::StepFunctionsTasks::EMRCreateCluster.new(self, "CreateCluster", {
    instances: {
        instance_fleets: [
            {
                instance_fleet_type: AWSCDK::StepFunctionsTasks::EMRCreateCluster::InstanceRoleType::CORE,
                instance_type_configs: [
                    {
                        instance_type: "m5.xlarge",
                    },
                ],
                target_on_demand_capacity: 1,
            },
            {
                instance_fleet_type: AWSCDK::StepFunctionsTasks::EMRCreateCluster::InstanceRoleType::MASTER,
                instance_type_configs: [
                    {
                        instance_type: "m5.xlarge",
                    },
                ],
                target_on_demand_capacity: 1,
            },
        ],
    },
    name: "ClusterName",
    release_label: "emr-7.9.0",
    managed_scaling_policy: {
        compute_limits: {
            unit_type: AWSCDK::StepFunctionsTasks::EMRCreateCluster::ComputeLimitsUnitType::INSTANCE_FLEET_UNITS,
            maximum_capacity_units: 4,
            minimum_capacity_units: 1,
            maximum_on_demand_capacity_units: 4,
            maximum_core_capacity_units: 2,
        },
    },
})

Termination Protection

Locks a cluster (job flow) so the EC2 instances in the cluster cannot be terminated by user intervention, an API call, or a job-flow error.

Corresponds to the set_termination_protection API in EMR.

AWSCDK::StepFunctionsTasks::EMRSetClusterTerminationProtection.new(self, "Task", {
    cluster_id: "ClusterId",
    termination_protected: false,
})

Terminate Cluster

Shuts down a cluster (job flow). Corresponds to the terminate_job_flows API in EMR.

AWSCDK::StepFunctionsTasks::EMRTerminateCluster.new(self, "Task", {
    cluster_id: "ClusterId",
})

Add Step

Adds a new step to a running cluster. Corresponds to the add_job_flow_steps API in EMR.

AWSCDK::StepFunctionsTasks::EMRAddStep.new(self, "Task", {
    cluster_id: "ClusterId",
    name: "StepName",
    jar: "Jar",
    action_on_failure: AWSCDK::StepFunctionsTasks::ActionOnFailure::CONTINUE,
})

To specify a custom runtime role use the execution_role_arn property.

Note: The EMR cluster must be created with a security configuration and the runtime role must have a specific trust policy. See this blog post for more details.

require 'aws-cdk-lib'


cfn_security_configuration = AWSCDK::EMR::CfnSecurityConfiguration.new(self, "EmrSecurityConfiguration", {
    name: "AddStepRuntimeRoleSecConfig",
    security_configuration: JSON[:parse](<<-'HERE'

        {
          "AuthorizationConfiguration": {
              "IAMConfiguration": {
                  "EnableApplicationScopedIAMRole": true,
                  "ApplicationScopedIAMRoleConfiguration":
                      {
                          "PropagateSourceIdentity": true
                      }
              },
              "LakeFormationConfiguration": {
                  "AuthorizedSessionTagValue": "Amazon EMR"
              }
          }
        }
    HERE),
})

task = AWSCDK::StepFunctionsTasks::EMRCreateCluster.new(self, "Create Cluster", {
    instances: {},
    name: AWSCDK::StepFunctions::TaskInput.from_json_path_at("$.ClusterName").value,
    security_configuration: cfn_security_configuration.name,
})

execution_role = AWSCDK::IAM::Role.new(self, "Role", {
    assumed_by: AWSCDK::IAM::ARNPrincipal.new(task.cluster_role.role_arn),
})

execution_role.assume_role_policy&.add_statements(
AWSCDK::IAM::PolicyStatement.new({
    effect: AWSCDK::IAM::Effect::ALLOW,
    principals: [
        task.cluster_role,
    ],
    actions: [
        "sts:SetSourceIdentity",
    ],
}),
AWSCDK::IAM::PolicyStatement.new({
    effect: AWSCDK::IAM::Effect::ALLOW,
    principals: [
        task.cluster_role,
    ],
    actions: [
        "sts:TagSession",
    ],
    conditions: {
        StringEquals: {
            "aws:RequestTag/LakeFormationAuthorizedCaller" => "Amazon EMR",
        },
    },
}))

AWSCDK::StepFunctionsTasks::EMRAddStep.new(self, "Task", {
    cluster_id: "ClusterId",
    execution_role_arn: execution_role.role_arn,
    name: "StepName",
    jar: "Jar",
    action_on_failure: AWSCDK::StepFunctionsTasks::ActionOnFailure::CONTINUE,
})

Cancel Step

Cancels a pending step in a running cluster. Corresponds to the cancel_steps API in EMR.

AWSCDK::StepFunctionsTasks::EMRCancelStep.new(self, "Task", {
    cluster_id: "ClusterId",
    step_id: "StepId",
})

Modify Instance Fleet

Modifies the target On-Demand and target Spot capacities for the instance fleet with the specified InstanceFleetName.

Corresponds to the modify_instance_fleet API in EMR.

AWSCDK::StepFunctionsTasks::EMRModifyInstanceFleetByName.new(self, "Task", {
    cluster_id: "ClusterId",
    instance_fleet_name: "InstanceFleetName",
    target_on_demand_capacity: 2,
    target_spot_capacity: 0,
})

Modify Instance Group

Modifies the number of nodes and configuration settings of an instance group.

Corresponds to the modify_instance_groups API in EMR.

AWSCDK::StepFunctionsTasks::EMRModifyInstanceGroupByName.new(self, "Task", {
    cluster_id: "ClusterId",
    instance_group_name: AWSCDK::StepFunctions::JsonPath.string_at("$.InstanceGroupName"),
    instance_group: {
        instance_count: 1,
    },
})

EMR on EKS

Step Functions supports Amazon EMR on EKS through the service integration pattern. The service integration APIs correspond to Amazon EMR on EKS APIs, but differ in the parameters that are used.

Read more about the differences when using these service integrations.

Setting up the EKS cluster is required.

Create Virtual Cluster

The CreateVirtualCluster API creates a single virtual cluster that's mapped to a single Kubernetes namespace.

The EKS cluster containing the Kubernetes namespace where the virtual cluster will be mapped can be passed in from the task input.

AWSCDK::StepFunctionsTasks::EMRContainersCreateVirtualCluster.new(self, "Create a Virtual Cluster", {
    eks_cluster: AWSCDK::StepFunctionsTasks::EKSClusterInput.from_task_input(AWSCDK::StepFunctions::TaskInput.from_text("clusterId")),
})

The EKS cluster can also be passed in directly.

require 'aws-cdk-lib'

eks_cluster = nil # AWSCDK::EKS::Cluster


AWSCDK::StepFunctionsTasks::EMRContainersCreateVirtualCluster.new(self, "Create a Virtual Cluster", {
    eks_cluster: AWSCDK::StepFunctionsTasks::EKSClusterInput.from_cluster(eks_cluster),
})

By default, the Kubernetes namespace that a virtual cluster maps to is "default", but a specific namespace within an EKS cluster can be selected.

AWSCDK::StepFunctionsTasks::EMRContainersCreateVirtualCluster.new(self, "Create a Virtual Cluster", {
    eks_cluster: AWSCDK::StepFunctionsTasks::EKSClusterInput.from_task_input(AWSCDK::StepFunctions::TaskInput.from_text("clusterId")),
    eks_namespace: "specified-namespace",
})

Delete Virtual Cluster

The DeleteVirtualCluster API deletes a virtual cluster.

AWSCDK::StepFunctionsTasks::EMRContainersDeleteVirtualCluster.new(self, "Delete a Virtual Cluster", {
    virtual_cluster_id: AWSCDK::StepFunctions::TaskInput.from_json_path_at("$.virtualCluster"),
})

Start Job Run

The StartJobRun API starts a job run. A job is a unit of work that you submit to Amazon EMR on EKS for execution. The work performed by the job can be defined by a Spark jar, PySpark script, or SparkSQL query. A job run is an execution of the job on the virtual cluster.

Required setup:

If not done already, follow the steps to setup EMR on EKS and create an EKS Cluster.
Enable Cluster access
Enable IAM Role access

The following actions must be performed if the virtual cluster ID is supplied from the task input. Otherwise, if it is supplied statically in the state machine definition, these actions will be done automatically.

Create an IAM role
Update the Role Trust Policy of the Job Execution Role.

The job can be configured with spark submit parameters:

AWSCDK::StepFunctionsTasks::EMRContainersStartJobRun.new(self, "EMR Containers Start Job Run", {
    virtual_cluster: AWSCDK::StepFunctionsTasks::VirtualClusterInput.from_virtual_cluster_id("de92jdei2910fwedz"),
    release_label: AWSCDK::StepFunctionsTasks::ReleaseLabel.EMR_6_2_0,
    job_driver: {
        spark_submit_job_driver: {
            entry_point: AWSCDK::StepFunctions::TaskInput.from_text("local:///usr/lib/spark/examples/src/main/python/pi.py"),
            spark_submit_parameters: "--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1",
        },
    },
})

Configuring the job can also be done via application configuration:

AWSCDK::StepFunctionsTasks::EMRContainersStartJobRun.new(self, "EMR Containers Start Job Run", {
    virtual_cluster: AWSCDK::StepFunctionsTasks::VirtualClusterInput.from_virtual_cluster_id("de92jdei2910fwedz"),
    release_label: AWSCDK::StepFunctionsTasks::ReleaseLabel.EMR_6_2_0,
    job_name: "EMR-Containers-Job",
    job_driver: {
        spark_submit_job_driver: {
            entry_point: AWSCDK::StepFunctions::TaskInput.from_text("local:///usr/lib/spark/examples/src/main/python/pi.py"),
        },
    },
    application_config: [
        {
            classification: AWSCDK::StepFunctionsTasks::Classification.SPARK_DEFAULTS,
            properties: {
                "spark.executor.instances" => "1",
                "spark.executor.memory" => "512M",
            },
        },
    ],
})

Job monitoring can be enabled if monitoring.logging is set true. This automatically generates an S3 bucket and CloudWatch logs.

AWSCDK::StepFunctionsTasks::EMRContainersStartJobRun.new(self, "EMR Containers Start Job Run", {
    virtual_cluster: AWSCDK::StepFunctionsTasks::VirtualClusterInput.from_virtual_cluster_id("de92jdei2910fwedz"),
    release_label: AWSCDK::StepFunctionsTasks::ReleaseLabel.EMR_6_2_0,
    job_driver: {
        spark_submit_job_driver: {
            entry_point: AWSCDK::StepFunctions::TaskInput.from_text("local:///usr/lib/spark/examples/src/main/python/pi.py"),
            spark_submit_parameters: "--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1",
        },
    },
    monitoring: {
        logging: true,
    },
})

Otherwise, providing monitoring for jobs with existing log groups and log buckets is also available.

require 'aws-cdk-lib'


log_group = AWSCDK::Logs::LogGroup.new(self, "Log Group")
log_bucket = AWSCDK::S3::Bucket.new(self, "S3 Bucket")

AWSCDK::StepFunctionsTasks::EMRContainersStartJobRun.new(self, "EMR Containers Start Job Run", {
    virtual_cluster: AWSCDK::StepFunctionsTasks::VirtualClusterInput.from_virtual_cluster_id("de92jdei2910fwedz"),
    release_label: AWSCDK::StepFunctionsTasks::ReleaseLabel.EMR_6_2_0,
    job_driver: {
        spark_submit_job_driver: {
            entry_point: AWSCDK::StepFunctions::TaskInput.from_text("local:///usr/lib/spark/examples/src/main/python/pi.py"),
            spark_submit_parameters: "--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1",
        },
    },
    monitoring: {
        log_group: log_group,
        log_bucket: log_bucket,
    },
})

Users can provide their own existing Job Execution Role.

AWSCDK::StepFunctionsTasks::EMRContainersStartJobRun.new(self, "EMR Containers Start Job Run", {
    virtual_cluster: AWSCDK::StepFunctionsTasks::VirtualClusterInput.from_task_input(AWSCDK::StepFunctions::TaskInput.from_json_path_at("$.VirtualClusterId")),
    release_label: AWSCDK::StepFunctionsTasks::ReleaseLabel.EMR_6_2_0,
    job_name: "EMR-Containers-Job",
    execution_role: AWSCDK::IAM::Role.from_role_arn(self, "Job-Execution-Role", "arn:aws:iam::xxxxxxxxxxxx:role/JobExecutionRole"),
    job_driver: {
        spark_submit_job_driver: {
            entry_point: AWSCDK::StepFunctions::TaskInput.from_text("local:///usr/lib/spark/examples/src/main/python/pi.py"),
            spark_submit_parameters: "--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1",
        },
    },
})

EKS

Step Functions supports Amazon EKS through the service integration pattern. The service integration APIs correspond to Amazon EKS APIs.

Read more about the differences when using these service integrations.

Call

Read and write Kubernetes resource objects via a Kubernetes API endpoint. Corresponds to the call API in Step Functions Connector.

The following code snippet includes a Task state that uses eks:call to list the pods.

require 'aws-cdk-lib'
require 'aws-cdk-lambda-layer-kubectl-v35'


my_eks_cluster = AWSCDK::EKS::Cluster.new(self, "my sample cluster", {
    version: AWSCDK::EKS::KubernetesVersion.V1_35,
    cluster_name: "myEksCluster",
    kubectl_layer: AWSCDK::LambdaLayerKubectlV35::KubectlV35Layer.new(self, "kubectl"),
})

AWSCDK::StepFunctionsTasks::EKSCall.new(self, "Call a EKS Endpoint", {
    cluster: my_eks_cluster,
    http_method: AWSCDK::StepFunctionsTasks::HttpMethods::GET,
    http_path: "/api/v1/namespaces/default/pods",
})

EventBridge

Step Functions supports Amazon EventBridge through the service integration pattern. The service integration APIs correspond to Amazon EventBridge APIs.

Read more about the differences when using these service integrations.

Put Events

Send events to an EventBridge bus. Corresponds to the put-events API in Step Functions Connector.

The following code snippet includes a Task state that uses events:putevents to send an event to the default bus.

require 'aws-cdk-lib'


my_event_bus = AWSCDK::Events::EventBus.new(self, "EventBus", {
    event_bus_name: "MyEventBus1",
})

AWSCDK::StepFunctionsTasks::EventBridgePutEvents.new(self, "Send an event to EventBridge", {
    entries: [
        {
            detail: AWSCDK::StepFunctions::TaskInput.from_object({
                Message: "Hello from Step Functions!",
            }),
            event_bus: my_event_bus,
            detail_type: "MessageFromStepFunctions",
            source: "step.functions",
        },
    ],
})

EventBridge Scheduler

You can call EventBridge Scheduler APIs from a Task state. Read more about calling Scheduler APIs here

Create Scheduler

The CreateSchedule API creates a new schedule.

Here is an example of how to create a schedule that puts an event to SQS queue every 5 minutes:

require 'aws-cdk-lib'

key = nil # AWSCDK::KMS::Key
schedule_group = nil # AWSCDK::Scheduler::CfnScheduleGroup
target_queue = nil # AWSCDK::SQS::Queue
dead_letter_queue = nil # AWSCDK::SQS::Queue


scheduler_role = AWSCDK::IAM::Role.new(self, "SchedulerRole", {
    assumed_by: AWSCDK::IAM::ServicePrincipal.new("scheduler.amazonaws.com"),
})
# To send the message to the queue
# This policy changes depending on the type of target.
scheduler_role.add_to_principal_policy(AWSCDK::IAM::PolicyStatement.new({
    actions: ["sqs:SendMessage"],
    resources: [target_queue.queue_arn],
}))

create_schedule_task1 = AWSCDK::StepFunctionsTasks::EventBridgeSchedulerCreateScheduleTask.new(self, "createSchedule", {
    schedule_name: "TestSchedule",
    action_after_completion: AWSCDK::StepFunctionsTasks::ActionAfterCompletion::NONE,
    client_token: "testToken",
    description: "TestDescription",
    start_date: Date.new,
    end_date: Date.new(Date.new[:get_time] + 1000 * 60 * 60),
    flexible_time_window: AWSCDK::Duration.minutes(5),
    group_name: schedule_group.ref,
    kms_key: key,
    schedule: AWSCDK::StepFunctionsTasks::Schedule.rate(AWSCDK::Duration.minutes(5)),
    timezone: "UTC",
    enabled: true,
    target: AWSCDK::StepFunctionsTasks::EventBridgeSchedulerTarget.new({
        arn: target_queue.queue_arn,
        role: scheduler_role,
        retry_policy: {
            maximum_retry_attempts: 2,
            maximum_event_age: AWSCDK::Duration.minutes(5),
        },
        dead_letter_queue: dead_letter_queue,
    }),
})

Glue

Step Functions supports AWS Glue through the service integration pattern.

StartJobRun

You can call the StartJobRun API from a Task state.

AWSCDK::StepFunctionsTasks::GlueStartJobRun.new(self, "Task", {
    glue_job_name: "my-glue-job",
    arguments: AWSCDK::StepFunctions::TaskInput.from_object({
        key: "value",
    }),
    task_timeout: AWSCDK::StepFunctions::Timeout.duration(AWSCDK::Duration.minutes(30)),
    notify_delay_after: AWSCDK::Duration.minutes(5),
})

You can configure workers by setting the worker_type_v2 and number_of_workers properties. worker_type is deprecated and no longer recommended. Use worker_type_v2 which is a ENUM-like class for more powerful worker configuration around using pre-defined values or dynamic values.

AWSCDK::StepFunctionsTasks::GlueStartJobRun.new(self, "Task", {
    glue_job_name: "my-glue-job",
    worker_configuration: {
        worker_type_v2: AWSCDK::StepFunctionsTasks::WorkerTypeV2.G_1X,
         # Worker type
        number_of_workers: 2,
    },
})

To configure the worker type or number of workers dynamically from StateMachine's input, you can configure it using JSON Path values using worker_type_v2 like this:

AWSCDK::StepFunctionsTasks::GlueStartJobRun.new(self, "Glue Job Task", {
    glue_job_name: "my-glue-job",
    worker_configuration: {
        worker_type_v2: AWSCDK::StepFunctionsTasks::WorkerTypeV2.of(AWSCDK::StepFunctions::JsonPath.string_at("$.glue_jobs_configs.executor_type")),
        number_of_workers: AWSCDK::StepFunctions::JsonPath.number_at("$.glue_jobs_configs.max_number_workers"),
    },
})

You can choose the execution class by setting the execution_class property.

AWSCDK::StepFunctionsTasks::GlueStartJobRun.new(self, "Task", {
    glue_job_name: "my-glue-job",
    execution_class: AWSCDK::StepFunctionsTasks::ExecutionClass::FLEX,
})

StartCrawlerRun

You can call the StartCrawler API from a Task state through AWS SDK service integrations.

require 'aws-cdk-lib'

my_crawler = nil # AWSCDK::Glue::CfnCrawler


# You can get the crawler name from `crawler.ref`
AWSCDK::StepFunctionsTasks::GlueStartCrawlerRun.new(self, "Task1", {
    crawler_name: my_crawler.ref,
})

# Of course, you can also specify the crawler name directly.
AWSCDK::StepFunctionsTasks::GlueStartCrawlerRun.new(self, "Task2", {
    crawler_name: "my-crawler-job",
})

Glue DataBrew

Step Functions supports AWS Glue DataBrew through the service integration pattern.

Start Job Run

You can call the StartJobRun API from a Task state.

AWSCDK::StepFunctionsTasks::GlueDataBrewStartJobRun.new(self, "Task", {
    name: "databrew-job",
})

Invoke HTTP API

Step Functions supports calling third-party APIs with credentials managed by Amazon EventBridge Connections.

The following snippet creates a new API destination connection, and uses it to make a POST request to the specified URL. The endpoint response is available at the $.ResponseBody path.

require 'aws-cdk-lib'


connection = AWSCDK::Events::Connection.new(self, "Connection", {
    authorization: AWSCDK::Events::Authorization.basic("username", AWSCDK::SecretValue.unsafe_plain_text("password")),
})

AWSCDK::StepFunctionsTasks::HttpInvoke.new(self, "Invoke HTTP API", {
    api_root: "https://api.example.com",
    api_endpoint: AWSCDK::StepFunctions::TaskInput.from_text(AWSCDK::StepFunctions::JsonPath.format("resource/{}/details", AWSCDK::StepFunctions::JsonPath.string_at("$.resourceId"))),
    body: AWSCDK::StepFunctions::TaskInput.from_object({foo: "bar"}),
    connection: connection,
    headers: AWSCDK::StepFunctions::TaskInput.from_object({"Content-Type" => "application/json"}),
    method: AWSCDK::StepFunctions::TaskInput.from_text("POST"),
    query_string_parameters: AWSCDK::StepFunctions::TaskInput.from_object({id: "123"}),
    url_encoding_format: AWSCDK::StepFunctionsTasks::URLEncodingFormat::BRACKETS,
})

Lambda

Step Functions supports AWS Lambda through the service integration pattern.

Invoke

Invoke a Lambda function.

You can specify the input to your Lambda function through the payload attribute. By default, Step Functions invokes Lambda function with the state input (JSON path '$') as the input.

The following snippet invokes a Lambda Function with the state input as the payload by referencing the $ path.

fn = nil # AWSCDK::Lambda::Function

AWSCDK::StepFunctionsTasks::LambdaInvoke.new(self, "Invoke with state input", {
    lambda_function: fn,
})

When a function is invoked, the Lambda service sends these response elements back.

⚠️ The response from the Lambda function is in an attribute called Payload

The following snippet invokes a Lambda Function by referencing the $.Payload path to reference the output of a Lambda executed before it.

fn = nil # AWSCDK::Lambda::Function

AWSCDK::StepFunctionsTasks::LambdaInvoke.new(self, "Invoke with empty object as payload", {
    lambda_function: fn,
    payload: AWSCDK::StepFunctions::TaskInput.from_object({}),
})

# use the output of fn as input
AWSCDK::StepFunctionsTasks::LambdaInvoke.new(self, "Invoke with payload field in the state input", {
    lambda_function: fn,
    payload: AWSCDK::StepFunctions::TaskInput.from_json_path_at("$.Payload"),
})

The following snippet invokes a Lambda and sets the task output to only include the Lambda function response.

fn = nil # AWSCDK::Lambda::Function

AWSCDK::StepFunctionsTasks::LambdaInvoke.new(self, "Invoke and set function response as task output", {
    lambda_function: fn,
    output_path: "$.Payload",
})

If you want to combine the input and the Lambda function response you can use the payload_response_only property and specify the result_path. This will put the Lambda function ARN directly in the "Resource" string, but it conflicts with the integrationPattern, invocationType, clientContext, and qualifier properties.

fn = nil # AWSCDK::Lambda::Function

AWSCDK::StepFunctionsTasks::LambdaInvoke.new(self, "Invoke and combine function response with task input", {
    lambda_function: fn,
    payload_response_only: true,
    result_path: "$.fn",
})

You can have Step Functions pause a task, and wait for an external process to return a task token. Read more about the callback pattern

To use the callback pattern, set the token property on the task. Call the Step Functions SendTaskSuccess or SendTaskFailure APIs with the token to indicate that the task has completed and the state machine should resume execution.

The following snippet invokes a Lambda with the task token as part of the input to the Lambda.

fn = nil # AWSCDK::Lambda::Function

AWSCDK::StepFunctionsTasks::LambdaInvoke.new(self, "Invoke with callback", {
    lambda_function: fn,
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::WAIT_FOR_TASK_TOKEN,
    payload: AWSCDK::StepFunctions::TaskInput.from_object({
        token: AWSCDK::StepFunctions::JsonPath.task_token,
        input: AWSCDK::StepFunctions::JsonPath.string_at("$.someField"),
    }),
})

⚠️ The task will pause until it receives that task token back with a SendTaskSuccess or SendTaskFailure call. Learn more about Callback with the Task Token.

AWS Lambda can occasionally experience transient service errors. In this case, invoking Lambda results in a 500 error, such as ClientExecutionTimeoutException, ServiceException, AWSLambdaException, or SdkClientException. As a best practice, the LambdaInvoke task will retry on those errors with an interval of 2 seconds, a back-off rate of 2 and 6 maximum attempts. Set the retry_on_service_exceptions prop to false to disable this behavior.

MediaConvert

Step Functions supports AWS MediaConvert through the Optimized integration pattern.

CreateJob

The CreateJob API creates a new transcoding job. For information about jobs and job settings, see the User Guide at https://docs.aws.amazon.com/mediaconvert/latest/ug/what-is.html

You can call the CreateJob API from a Task state. Optionally you can specify the integration_pattern.

Make sure you update the required fields - Role & Settings and refer CreateJobRequest for all other optional parameters.

AWSCDK::StepFunctionsTasks::MediaConvertCreateJob.new(self, "CreateJob", {
    create_job_request: {
        "Role" => "arn:aws:iam::123456789012:role/MediaConvertRole",
        "Settings" => {
            "OutputGroups" => [
                {
                    "Outputs" => [
                        {
                            "ContainerSettings" => {
                                "Container" => "MP4",
                            },
                            "VideoDescription" => {
                                "CodecSettings" => {
                                    "Codec" => "H_264",
                                    "H264Settings" => {
                                        "MaxBitrate" => 1000,
                                        "RateControlMode" => "QVBR",
                                        "SceneChangeDetect" => "TRANSITION_DETECTION",
                                    },
                                },
                            },
                            "AudioDescriptions" => [
                                {
                                    "CodecSettings" => {
                                        "Codec" => "AAC",
                                        "AacSettings" => {
                                            "Bitrate" => 96000,
                                            "CodingMode" => "CODING_MODE_2_0",
                                            "SampleRate" => 48000,
                                        },
                                    },
                                },
                            ],
                        },
                    ],
                    "OutputGroupSettings" => {
                        "Type" => "FILE_GROUP_SETTINGS",
                        "FileGroupSettings" => {
                            "Destination" => "s3://EXAMPLE-DESTINATION-BUCKET/",
                        },
                    },
                },
            ],
            "Inputs" => [
                {
                    "AudioSelectors" => {
                        "Audio Selector 1" => {
                            "DefaultSelection" => "DEFAULT",
                        },
                    },
                    "FileInput" => "s3://EXAMPLE-SOURCE-BUCKET/EXAMPLE-SOURCE_FILE",
                },
            ],
        },
    },
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::RUN_JOB,
})

SageMaker

Step Functions supports AWS SageMaker through the service integration pattern.

If your training job or model uses resources from AWS Marketplace, network isolation is required. To do so, set the enable_network_isolation property to true for SageMakerCreateModel or SageMakerCreateTrainingJob.

To set environment variables for the Docker container use the environment property.

Create Training Job

You can call the CreateTrainingJob API from a Task state.

AWSCDK::StepFunctionsTasks::SageMakerCreateTrainingJob.new(self, "TrainSagemaker", {
    training_job_name: AWSCDK::StepFunctions::JsonPath.string_at("$.JobName"),
    algorithm_specification: {
        algorithm_name: "BlazingText",
        training_input_mode: AWSCDK::StepFunctionsTasks::InputMode::FILE,
    },
    input_data_config: [
        {
            channel_name: "train",
            data_source: {
                s3_data_source: {
                    s3_data_type: AWSCDK::StepFunctionsTasks::S3DataType::S3_PREFIX,
                    s3_location: AWSCDK::StepFunctionsTasks::S3Location.from_json_expression("$.S3Bucket"),
                },
            },
        },
    ],
    output_data_config: {
        s3_output_location: AWSCDK::StepFunctionsTasks::S3Location.from_bucket(AWSCDK::S3::Bucket.from_bucket_name(self, "Bucket", "amzn-s3-demo-bucket"), "myoutputpath"),
    },
    resource_config: {
        instance_count: 1,
        instance_type: AWSCDK::EC2::InstanceType.new(AWSCDK::StepFunctions::JsonPath.string_at("$.InstanceType")),
        volume_size: AWSCDK::Size.gibibytes(50),
    },
     # optional: default is 1 instance of EC2 `M4.XLarge` with `10GB` volume
    stopping_condition: {
        max_runtime: AWSCDK::Duration.hours(2),
    },
})

You can specify TrainingInputMode via the trainingInputMode property.

To download the data from Amazon Simple Storage Service (Amazon S3) to the provisioned ML storage volume, and mount the directory to a Docker volume, choose InputMode.FILE if an algorithm supports it.
To stream data directly from Amazon S3 to the container, choose InputMode.PIPE if an algorithm supports it.
To stream data directly from Amazon S3 to the container with no code changes and to provide file system access to the data, choose InputMode.FAST_FILE if an algorithm supports it.

Create Transform Job

You can call the CreateTransformJob API from a Task state.

AWSCDK::StepFunctionsTasks::SageMakerCreateTransformJob.new(self, "Batch Inference", {
    transform_job_name: "MyTransformJob",
    model_name: "MyModelName",
    model_client_options: {
        invocations_max_retries: 3,
         # default is 0
        invocations_timeout: AWSCDK::Duration.minutes(5),
    },
    transform_input: {
        transform_data_source: {
            s3_data_source: {
                s3_uri: "s3://inputbucket/train",
                s3_data_type: AWSCDK::StepFunctionsTasks::S3DataType::S3_PREFIX,
            },
        },
    },
    transform_output: {
        s3_output_path: "s3://outputbucket/TransformJobOutputPath",
    },
    transform_resources: {
        instance_count: 1,
        instance_type: AWSCDK::EC2::InstanceType.of(AWSCDK::EC2::InstanceClass::M4, AWSCDK::EC2::InstanceSize::XLARGE),
    },
})

Create Endpoint

You can call the CreateEndpoint API from a Task state.

AWSCDK::StepFunctionsTasks::SageMakerCreateEndpoint.new(self, "SagemakerEndpoint", {
    endpoint_name: AWSCDK::StepFunctions::JsonPath.string_at("$.EndpointName"),
    endpoint_config_name: AWSCDK::StepFunctions::JsonPath.string_at("$.EndpointConfigName"),
})

Create Endpoint Config

You can call the CreateEndpointConfig API from a Task state.

AWSCDK::StepFunctionsTasks::SageMakerCreateEndpointConfig.new(self, "SagemakerEndpointConfig", {
    endpoint_config_name: "MyEndpointConfig",
    production_variants: [
        {
            initial_instance_count: 2,
            instance_type: AWSCDK::EC2::InstanceType.of(AWSCDK::EC2::InstanceClass::M5, AWSCDK::EC2::InstanceSize::XLARGE),
            model_name: "MyModel",
            variant_name: "awesome-variant",
        },
    ],
})

Create Model

You can call the CreateModel API from a Task state.

AWSCDK::StepFunctionsTasks::SageMakerCreateModel.new(self, "Sagemaker", {
    model_name: "MyModel",
    primary_container: AWSCDK::StepFunctionsTasks::ContainerDefinition.new({
        image: AWSCDK::StepFunctionsTasks::DockerImage.from_json_expression(AWSCDK::StepFunctions::JsonPath.string_at("$.Model.imageName")),
        mode: AWSCDK::StepFunctionsTasks::Mode::SINGLE_MODEL,
        model_s3_location: AWSCDK::StepFunctionsTasks::S3Location.from_json_expression("$.TrainingJob.ModelArtifacts.S3ModelArtifacts"),
    }),
})

Update Endpoint

You can call the UpdateEndpoint API from a Task state.

AWSCDK::StepFunctionsTasks::SageMakerUpdateEndpoint.new(self, "SagemakerEndpoint", {
    endpoint_name: AWSCDK::StepFunctions::JsonPath.string_at("$.Endpoint.Name"),
    endpoint_config_name: AWSCDK::StepFunctions::JsonPath.string_at("$.Endpoint.EndpointConfig"),
})

SNS

Step Functions supports Amazon SNS through the service integration pattern.

Publish

You can call the Publish API from a Task state to publish to an SNS topic.

topic = AWSCDK::SNS::Topic.new(self, "Topic")

# Use a field from the execution data as message.
task1 = AWSCDK::StepFunctionsTasks::SNSPublish.new(self, "Publish1", {
    topic: topic,
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::REQUEST_RESPONSE,
    message: AWSCDK::StepFunctions::TaskInput.from_data_at("$.state.message"),
    message_attributes: {
        place: {
            value: AWSCDK::StepFunctions::JsonPath.string_at("$.place"),
        },
        pic: {
            # BINARY must be explicitly set
            data_type: AWSCDK::StepFunctionsTasks::MessageAttributeDataType::BINARY,
            value: AWSCDK::StepFunctions::JsonPath.string_at("$.pic"),
        },
        people: {
            value: 4,
        },
        handles: {
            value: ["@kslater", "@jjf", nil, "@mfanning"],
        },
    },
})

# Combine a field from the execution data with
# a literal object.
task2 = AWSCDK::StepFunctionsTasks::SNSPublish.new(self, "Publish2", {
    topic: topic,
    message: AWSCDK::StepFunctions::TaskInput.from_object({
        field1: "somedata",
        field2: AWSCDK::StepFunctions::JsonPath.string_at("$.field2"),
    }),
})

Step Functions

Step Functions supports AWS Step Functions through the service integration pattern.

Start Execution

You can manage AWS Step Functions executions.

AWS Step Functions supports its own StartExecution API as a service integration.

# Define a state machine with one Pass state
child = AWSCDK::StepFunctions::StateMachine.new(self, "ChildStateMachine", {
    definition: AWSCDK::StepFunctions::Chain.start(AWSCDK::StepFunctions::Pass.new(self, "PassState")),
})

# Include the state machine in a Task state with callback pattern
task = AWSCDK::StepFunctionsTasks::StepFunctionsStartExecution.new(self, "ChildTask", {
    state_machine: child,
    integration_pattern: AWSCDK::StepFunctions::IntegrationPattern::WAIT_FOR_TASK_TOKEN,
    input: AWSCDK::StepFunctions::TaskInput.from_object({
        token: AWSCDK::StepFunctions::JsonPath.task_token,
        foo: "bar",
    }),
    name: "MyExecutionName",
})

# Define a second state machine with the Task state above
AWSCDK::StepFunctions::StateMachine.new(self, "ParentStateMachine", {
    definition: task,
})

You can utilize Associate Workflow Executions via the associate_with_parent property. This allows the Step Functions UI to link child executions from parent executions, making it easier to trace execution flow across state machines.

child = nil # AWSCDK::StepFunctions::StateMachine

task = AWSCDK::StepFunctionsTasks::StepFunctionsStartExecution.new(self, "ChildTask", {
    state_machine: child,
    associate_with_parent: true,
})

This will add the payload AWS_STEP_FUNCTIONS_STARTED_BY_EXECUTION_ID.$: $$.Execution.Id to the inputproperty for you, which will pass the execution ID from the context object to the execution input. It requires input to be an object or not be set at all.

Invoke Activity

You can invoke a Step Functions Activity which enables you to have a task in your state machine where the work is performed by a worker that can be hosted on Amazon EC2, Amazon ECS, AWS Lambda, basically anywhere. Activities are a way to associate code running somewhere (known as an activity worker) with a specific task in a state machine.

When Step Functions reaches an activity task state, the workflow waits for an activity worker to poll for a task. An activity worker polls Step Functions by using GetActivityTask, and sending the ARN for the related activity.

After the activity worker completes its work, it can provide a report of its success or failure by using SendTaskSuccess or SendTaskFailure. These two calls use the taskToken provided by GetActivityTask to associate the result with that task.

The following example creates an activity and creates a task that invokes the activity.

submit_job_activity = AWSCDK::StepFunctions::Activity.new(self, "SubmitJob")

AWSCDK::StepFunctionsTasks::StepFunctionsInvokeActivity.new(self, "Submit Job", {
    activity: submit_job_activity,
})

Use the Parameters field to create a collection of key-value pairs that are passed as input. The values of each can either be static values that you include in your state machine definition, or selected from either the input or the context object with a path.

submit_job_activity = AWSCDK::StepFunctions::Activity.new(self, "SubmitJob")

AWSCDK::StepFunctionsTasks::StepFunctionsInvokeActivity.new(self, "Submit Job", {
    activity: submit_job_activity,
    parameters: {
        comment: "Selecting what I care about.",
        MyDetails: {
            size: AWSCDK::StepFunctions::JsonPath.string_at("$.product.details.size"),
            exists: AWSCDK::StepFunctions::JsonPath.string_at("$.product.availability"),
            StaticValue: "foo",
        },
    },
})

SQS

Step Functions supports Amazon SQS

Send Message

You can call the SendMessage API from a Task state to send a message to an SQS queue.

queue = AWSCDK::SQS::Queue.new(self, "Queue")

# Use a field from the execution data as message.
task1 = AWSCDK::StepFunctionsTasks::SQSSendMessage.new(self, "Send1", {
    queue: queue,
    message_body: AWSCDK::StepFunctions::TaskInput.from_json_path_at("$.message"),
})

# Combine a field from the execution data with
# a literal object.
task2 = AWSCDK::StepFunctionsTasks::SQSSendMessage.new(self, "Send2", {
    queue: queue,
    message_body: AWSCDK::StepFunctions::TaskInput.from_object({
        field1: "somedata",
        field2: AWSCDK::StepFunctions::JsonPath.string_at("$.field2"),
    }),
})

API Reference

Classes 66

Interfaces 203

Enums 24

ActionAfterCompletionThe action that EventBridge Scheduler applies to the schedule after the schedule completes ActionOnFailureThe action to take when the cluster step fails. AssembleWithHow to assemble the results of the transform job as a single S3 object. AuthTypeThe authentication method used to call the endpoint. BatchStrategySpecifies the number of records to include in a mini-batch for an HTTP inference request. CompressionTypeCompression type of the data. CustomizationTypeThe customization type. DynamoConsumedCapacityDetermines the level of detail about provisioned throughput consumption that is returned. DynamoItemCollectionMetricsDetermines whether item collection metrics are returned. DynamoReturnValuesUse ReturnValues if you want to get the item attributes as they appear before or after the EncryptionOptionEncryption Options of the S3 bucket. ExecutionClassThe excecution class of the job. HttpMethodHttp Methods that API Gateway supports. HttpMethodsMethod type of a EKS call. InputModeInput mode that the algorithm supports. LambdaInvocationTypeInvocation type of a Lambda. MessageAttributeDataTypeThe data type set for the SNS message attributes. ModeSpecifies how many models the container hosts. RecordWrapperTypeDefine the format of the input data. S3DataDistributionTypeS3 Data Distribution Type. S3DataTypeS3 Data Type. SplitTypeMethod to use to split the transform job's data files into smaller batches. URLEncodingFormatThe style used when applying URL encoding to array values. WorkerTypeThe type of predefined worker that is allocated when a job runs.

AWSCDK::StepFunctionsTasks

Tasks for AWS Step Functions

Table Of Contents

Paths

Evaluate Expression

API Gateway

Call REST API Endpoint

Call HTTP API Endpoint

AWS SDK

Cross-region AWS API call

Athena

StartQueryExecution

GetQueryExecution

GetQueryResults

StopQueryExecution

Batch

SubmitJob

Bedrock

InvokeModel

Using Input Path for S3 URI

Using Input Path

createModelCustomizationJob

CodeBuild

StartBuild

StartBuildBatch

DynamoDB

GetItem

PutItem

DeleteItem

UpdateItem

ECS

RunTask

EC2

Fargate

Capacity Provider Options

Override CPU and Memory Parameter

ECS enable Exec

EMR

Create Cluster

Termination Protection

Terminate Cluster

Add Step

Cancel Step

Modify Instance Fleet

Modify Instance Group

EMR on EKS

Create Virtual Cluster

Delete Virtual Cluster

Start Job Run

EKS

Call

EventBridge

Put Events

EventBridge Scheduler

Create Scheduler

Glue

StartJobRun

StartCrawlerRun

Glue DataBrew

Start Job Run

Invoke HTTP API

Lambda

Invoke

MediaConvert

CreateJob

SageMaker

Create Training Job

Create Transform Job

Create Endpoint

Create Endpoint Config

Create Model

Update Endpoint

SNS

Publish

Step Functions

Start Execution

Invoke Activity

SQS

Send Message

API Reference