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.
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 Node.js Lambda function. 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:
ts
const convertToSeconds = new tasks.EvaluateExpression(this, 'Convert to seconds', {
expression: '$.waitMilliseconds / 1000',
resultPath: '$.waitSeconds',
});
const createMessage = new tasks.EvaluateExpression(this, 'Create message', {
// Note: this is a string inside a string.
expression: '`Now waiting ${$.waitSeconds} seconds...`',
runtime: lambda.Runtime.NODEJS_LATEST,
resultPath: '$.message',
});
const publishMessage = new tasks.SnsPublish(this, 'Publish message', {
topic: new sns.Topic(this, 'cool-topic'),
message: sfn.TaskInput.fromJsonPathAt('$.message'),
resultPath: '$.sns',
});
const wait = new sfn.Wait(this, 'Wait', {
time: sfn.WaitTime.secondsPath('$.waitSeconds'),
});
new sfn.StateMachine(this, 'StateMachine', {
definition: convertToSeconds
.next(createMessage)
.next(publishMessage)
.next(wait),
});
python
convert_to_seconds = tasks.EvaluateExpression(self, "Convert to seconds",
expression="$.waitMilliseconds / 1000",
result_path="$.waitSeconds"
)
create_message = tasks.EvaluateExpression(self, "Create message",
# Note: this is a string inside a string.
expression="`Now waiting ${$.waitSeconds} seconds...`",
runtime=lambda_.Runtime.NODEJS_LATEST,
result_path="$.message"
)
publish_message = tasks.SnsPublish(self, "Publish message",
topic=sns.Topic(self, "cool-topic"),
message=sfn.TaskInput.from_json_path_at("$.message"),
result_path="$.sns"
)
wait = sfn.Wait(self, "Wait",
time=sfn.WaitTime.seconds_path("$.waitSeconds")
)
sfn.StateMachine(self, "StateMachine",
definition=convert_to_seconds.next(create_message).next(publish_message).next(wait)
)
java
EvaluateExpression convertToSeconds = EvaluateExpression.Builder.create(this, "Convert to seconds")
.expression("$.waitMilliseconds / 1000")
.resultPath("$.waitSeconds")
.build();
EvaluateExpression createMessage = EvaluateExpression.Builder.create(this, "Create message")
// Note: this is a string inside a string.
.expression("`Now waiting ${$.waitSeconds} seconds...`")
.runtime(Runtime.NODEJS_LATEST)
.resultPath("$.message")
.build();
SnsPublish publishMessage = SnsPublish.Builder.create(this, "Publish message")
.topic(new Topic(this, "cool-topic"))
.message(TaskInput.fromJsonPathAt("$.message"))
.resultPath("$.sns")
.build();
Wait wait = Wait.Builder.create(this, "Wait")
.time(WaitTime.secondsPath("$.waitSeconds"))
.build();
StateMachine.Builder.create(this, "StateMachine")
.definition(convertToSeconds.next(createMessage).next(publishMessage).next(wait))
.build();
csharp
var convertToSeconds = new EvaluateExpression(this, "Convert to seconds", new EvaluateExpressionProps {
Expression = "$.waitMilliseconds / 1000",
ResultPath = "$.waitSeconds"
});
var createMessage = new EvaluateExpression(this, "Create message", new EvaluateExpressionProps {
// Note: this is a string inside a string.
Expression = "`Now waiting ${$.waitSeconds} seconds...`",
Runtime = Runtime.NODEJS_LATEST,
ResultPath = "$.message"
});
var publishMessage = new SnsPublish(this, "Publish message", new SnsPublishProps {
Topic = new Topic(this, "cool-topic"),
Message = TaskInput.FromJsonPathAt("$.message"),
ResultPath = "$.sns"
});
var wait = new Wait(this, "Wait", new WaitProps {
Time = WaitTime.SecondsPath("$.waitSeconds")
});
new StateMachine(this, "StateMachine", new StateMachineProps {
Definition = convertToSeconds.Next(createMessage).Next(publishMessage).Next(wait)
});
go
convertToSeconds := tasks.NewEvaluateExpression(this, jsii.String("Convert to seconds"), &EvaluateExpressionProps{
Expression: jsii.String("$.waitMilliseconds / 1000"),
ResultPath: jsii.String("$.waitSeconds"),
})
createMessage := tasks.NewEvaluateExpression(this, jsii.String("Create message"), &EvaluateExpressionProps{
// Note: this is a string inside a string.
Expression: jsii.String("`Now waiting ${$.waitSeconds} seconds...`"),
Runtime: lambda.Runtime_NODEJS_LATEST(),
ResultPath: jsii.String("$.message"),
})
publishMessage := tasks.NewSnsPublish(this, jsii.String("Publish message"), &SnsPublishProps{
Topic: sns.NewTopic(this, jsii.String("cool-topic")),
Message: sfn.TaskInput_FromJsonPathAt(jsii.String("$.message")),
ResultPath: jsii.String("$.sns"),
})
wait := sfn.NewWait(this, jsii.String("Wait"), &WaitProps{
Time: sfn.WaitTime_SecondsPath(jsii.String("$.waitSeconds")),
})
sfn.NewStateMachine(this, jsii.String("StateMachine"), &StateMachineProps{
Definition: convertToSeconds.Next(createMessage).Next(publishMessage).*Next(wait),
})
Referenced state paths are only resolved where they are used as code (for example $.a + $.b, or a function argument). To embed a value inside a string, use a template literal interpolation as in the Create message example above (${$.waitSeconds}). A path written as literal text inside a plain string, or as bare text inside a template literal (for example 'waited $.waitSeconds seconds'), is not resolved and is rejected at synthesis time.
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.
ts
const convertToSecondsArm64 = new tasks.EvaluateExpression(this, 'Convert to seconds', {
expression: '$.waitMilliseconds / 1000',
architecture: lambda.Architecture.ARM_64,
});
python
convert_to_seconds_arm64 = tasks.EvaluateExpression(self, "Convert to seconds",
expression="$.waitMilliseconds / 1000",
architecture=lambda_.Architecture.ARM_64
)
java
EvaluateExpression convertToSecondsArm64 = EvaluateExpression.Builder.create(this, "Convert to seconds")
.expression("$.waitMilliseconds / 1000")
.architecture(Architecture.ARM_64)
.build();
csharp
var convertToSecondsArm64 = new EvaluateExpression(this, "Convert to seconds", new EvaluateExpressionProps {
Expression = "$.waitMilliseconds / 1000",
Architecture = Architecture.ARM_64
});
go
convertToSecondsArm64 := tasks.NewEvaluateExpression(this, jsii.String("Convert to seconds"), &EvaluateExpressionProps{
Expression: jsii.String("$.waitMilliseconds / 1000"),
Architecture: 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.
ts
import * as apigateway from 'aws-cdk-lib/aws-apigateway';
const restApi = new apigateway.RestApi(this, 'MyRestApi');
const invokeTask = new tasks.CallApiGatewayRestApiEndpoint(this, 'Call REST API', {
api: restApi,
stageName: 'prod',
method: tasks.HttpMethod.GET,
});
python
import aws_cdk.aws_apigateway as apigateway
rest_api = apigateway.RestApi(self, "MyRestApi")
invoke_task = tasks.CallApiGatewayRestApiEndpoint(self, "Call REST API",
api=rest_api,
stage_name="prod",
method=tasks.HttpMethod.GET
)
java
import software.amazon.awscdk.services.apigateway.*;
RestApi restApi = new RestApi(this, "MyRestApi");
CallApiGatewayRestApiEndpoint invokeTask = CallApiGatewayRestApiEndpoint.Builder.create(this, "Call REST API")
.api(restApi)
.stageName("prod")
.method(HttpMethod.GET)
.build();
csharp
using Amazon.CDK.AWS.APIGateway;
var restApi = new RestApi(this, "MyRestApi");
var invokeTask = new CallApiGatewayRestApiEndpoint(this, "Call REST API", new CallApiGatewayRestApiEndpointProps {
Api = restApi,
StageName = "prod",
Method = HttpMethod.GET
});
go
import apigateway "github.com/aws/aws-cdk-go/awscdk"
restApi := apigateway.NewRestApi(this, jsii.String("MyRestApi"))
invokeTask := tasks.NewCallApiGatewayRestApiEndpoint(this, jsii.String("Call REST API"), &CallApiGatewayRestApiEndpointProps{
Api: restApi,
StageName: jsii.String("prod"),
Method: tasks.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:
ts
import * as apigateway from 'aws-cdk-lib/aws-apigateway';
const restApi = new apigateway.RestApi(this, 'MyRestApi');
const invokeTask = new tasks.CallApiGatewayRestApiEndpoint(this, 'Call REST API', {
api: restApi,
stageName: 'prod',
method: tasks.HttpMethod.GET,
region: 'us-west-2',
});
python
import aws_cdk.aws_apigateway as apigateway
rest_api = apigateway.RestApi(self, "MyRestApi")
invoke_task = tasks.CallApiGatewayRestApiEndpoint(self, "Call REST API",
api=rest_api,
stage_name="prod",
method=tasks.HttpMethod.GET,
region="us-west-2"
)
java
import software.amazon.awscdk.services.apigateway.*;
RestApi restApi = new RestApi(this, "MyRestApi");
CallApiGatewayRestApiEndpoint invokeTask = CallApiGatewayRestApiEndpoint.Builder.create(this, "Call REST API")
.api(restApi)
.stageName("prod")
.method(HttpMethod.GET)
.region("us-west-2")
.build();
csharp
using Amazon.CDK.AWS.APIGateway;
var restApi = new RestApi(this, "MyRestApi");
var invokeTask = new CallApiGatewayRestApiEndpoint(this, "Call REST API", new CallApiGatewayRestApiEndpointProps {
Api = restApi,
StageName = "prod",
Method = HttpMethod.GET,
Region = "us-west-2"
});
go
import apigateway "github.com/aws/aws-cdk-go/awscdk"
restApi := apigateway.NewRestApi(this, jsii.String("MyRestApi"))
invokeTask := tasks.NewCallApiGatewayRestApiEndpoint(this, jsii.String("Call REST API"), &CallApiGatewayRestApiEndpointProps{
Api: restApi,
StageName: jsii.String("prod"),
Method: tasks.HttpMethod_GET,
Region: jsii.String("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:
ts
import * as apigateway from 'aws-cdk-lib/aws-apigateway';
declare const api: apigateway.RestApi;
new tasks.CallApiGatewayRestApiEndpoint(this, 'Endpoint', {
api,
stageName: 'Stage',
method: tasks.HttpMethod.PUT,
integrationPattern: sfn.IntegrationPattern.WAIT_FOR_TASK_TOKEN,
headers: sfn.TaskInput.fromObject({
TaskToken: sfn.JsonPath.array(sfn.JsonPath.taskToken),
}),
});
python
import aws_cdk.aws_apigateway as apigateway
# api: apigateway.RestApi
tasks.CallApiGatewayRestApiEndpoint(self, "Endpoint",
api=api,
stage_name="Stage",
method=tasks.HttpMethod.PUT,
integration_pattern=sfn.IntegrationPattern.WAIT_FOR_TASK_TOKEN,
headers=sfn.TaskInput.from_object({
"TaskToken": sfn.JsonPath.array(sfn.JsonPath.task_token)
})
)
java
import software.amazon.awscdk.services.apigateway.*;
RestApi api;
CallApiGatewayRestApiEndpoint.Builder.create(this, "Endpoint")
.api(api)
.stageName("Stage")
.method(HttpMethod.PUT)
.integrationPattern(IntegrationPattern.WAIT_FOR_TASK_TOKEN)
.headers(TaskInput.fromObject(Map.of(
"TaskToken", JsonPath.array(JsonPath.getTaskToken()))))
.build();
csharp
using Amazon.CDK.AWS.APIGateway;
RestApi api;
new CallApiGatewayRestApiEndpoint(this, "Endpoint", new CallApiGatewayRestApiEndpointProps {
Api = api,
StageName = "Stage",
Method = HttpMethod.PUT,
IntegrationPattern = IntegrationPattern.WAIT_FOR_TASK_TOKEN,
Headers = TaskInput.FromObject(new Dictionary<string, object> {
{ "TaskToken", JsonPath.Array(JsonPath.TaskToken) }
})
});
go
import apigateway "github.com/aws/aws-cdk-go/awscdk"
var api RestApi
tasks.NewCallApiGatewayRestApiEndpoint(this, jsii.String("Endpoint"), &CallApiGatewayRestApiEndpointProps{
Api: Api,
StageName: jsii.String("Stage"),
Method: tasks.HttpMethod_PUT,
IntegrationPattern: sfn.IntegrationPattern_WAIT_FOR_TASK_TOKEN,
Headers: sfn.TaskInput_FromObject(map[string]interface{}{
"TaskToken": sfn.JsonPath_array(sfn.JsonPath_taskToken()),
}),
})
Call HTTP API Endpoint
The CallApiGatewayHttpApiEndpoint calls the HTTP API endpoint.
ts
import * as apigatewayv2 from 'aws-cdk-lib/aws-apigatewayv2';
const httpApi = new apigatewayv2.HttpApi(this, 'MyHttpApi');
const invokeTask = new tasks.CallApiGatewayHttpApiEndpoint(this, 'Call HTTP API', {
apiId: httpApi.apiId,
apiStack: Stack.of(httpApi),
method: tasks.HttpMethod.GET,
});
python
import aws_cdk.aws_apigatewayv2 as apigatewayv2
http_api = apigatewayv2.HttpApi(self, "MyHttpApi")
invoke_task = tasks.CallApiGatewayHttpApiEndpoint(self, "Call HTTP API",
api_id=http_api.api_id,
api_stack=Stack.of(http_api),
method=tasks.HttpMethod.GET
)
java
import software.amazon.awscdk.services.apigatewayv2.*;
HttpApi httpApi = new HttpApi(this, "MyHttpApi");
CallApiGatewayHttpApiEndpoint invokeTask = CallApiGatewayHttpApiEndpoint.Builder.create(this, "Call HTTP API")
.apiId(httpApi.getApiId())
.apiStack(Stack.of(httpApi))
.method(HttpMethod.GET)
.build();
csharp
using Amazon.CDK.AWS.Apigatewayv2;
var httpApi = new HttpApi(this, "MyHttpApi");
var invokeTask = new CallApiGatewayHttpApiEndpoint(this, "Call HTTP API", new CallApiGatewayHttpApiEndpointProps {
ApiId = httpApi.ApiId,
ApiStack = Stack.Of(httpApi),
Method = HttpMethod.GET
});
go
import apigatewayv2 "github.com/aws/aws-cdk-go/awscdk"
httpApi := apigatewayv2.NewHttpApi(this, jsii.String("MyHttpApi"))
invokeTask := tasks.NewCallApiGatewayHttpApiEndpoint(this, jsii.String("Call HTTP API"), &CallApiGatewayHttpApiEndpointProps{
ApiId: httpApi.ApiId,
ApiStack: awscdk.*stack_Of(httpApi),
Method: tasks.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.
ts
declare const myBucket: s3.Bucket;
const getObject = new tasks.CallAwsService(this, 'GetObject', {
service: 's3',
action: 'getObject',
parameters: {
Bucket: myBucket.bucketName,
Key: sfn.JsonPath.stringAt('$.key')
},
iamResources: [myBucket.arnForObjects('*')],
});
python
# my_bucket: s3.Bucket
get_object = tasks.CallAwsService(self, "GetObject",
service="s3",
action="getObject",
parameters={
"Bucket": my_bucket.bucket_name,
"Key": sfn.JsonPath.string_at("$.key")
},
iam_resources=[my_bucket.arn_for_objects("*")]
)
java
Bucket myBucket;
CallAwsService getObject = CallAwsService.Builder.create(this, "GetObject")
.service("s3")
.action("getObject")
.parameters(Map.of(
"Bucket", myBucket.getBucketName(),
"Key", JsonPath.stringAt("$.key")))
.iamResources(List.of(myBucket.arnForObjects("*")))
.build();
csharp
Bucket myBucket;
var getObject = new CallAwsService(this, "GetObject", new CallAwsServiceProps {
Service = "s3",
Action = "getObject",
Parameters = new Dictionary<string, object> {
{ "Bucket", myBucket.BucketName },
{ "Key", JsonPath.StringAt("$.key") }
},
IamResources = new [] { myBucket.ArnForObjects("*") }
});
go
var myBucket Bucket
getObject := tasks.NewCallAwsService(this, jsii.String("GetObject"), &CallAwsServiceProps{
Service: jsii.String("s3"),
Action: jsii.String("getObject"),
Parameters: map[string]interface{}{
"Bucket": myBucket.bucketName,
"Key": sfn.JsonPath_stringAt(jsii.String("$.key")),
},
IamResources: []*string{
myBucket.ArnForObjects(jsii.String("*")),
},
})
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 iamResources.
Use the iamAction 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:
ts
const listBuckets = new tasks.CallAwsService(this, 'ListBuckets', {
service: 's3',
action: 'listBuckets',
iamResources: ['*'],
iamAction: 's3:ListAllMyBuckets',
});
python
list_buckets = tasks.CallAwsService(self, "ListBuckets",
service="s3",
action="listBuckets",
iam_resources=["*"],
iam_action="s3:ListAllMyBuckets"
)
java
CallAwsService listBuckets = CallAwsService.Builder.create(this, "ListBuckets")
.service("s3")
.action("listBuckets")
.iamResources(List.of("*"))
.iamAction("s3:ListAllMyBuckets")
.build();
csharp
var listBuckets = new CallAwsService(this, "ListBuckets", new CallAwsServiceProps {
Service = "s3",
Action = "listBuckets",
IamResources = new [] { "*" },
IamAction = "s3:ListAllMyBuckets"
});
go
listBuckets := tasks.NewCallAwsService(this, jsii.String("ListBuckets"), &CallAwsServiceProps{
Service: jsii.String("s3"),
Action: jsii.String("listBuckets"),
IamResources: []*string{
jsii.String("*"),
},
IamAction: jsii.String("s3:ListAllMyBuckets"),
})
Use the additionalIamStatements 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:
ts
const detectLabels = new tasks.CallAwsService(this, 'DetectLabels', {
service: 'rekognition',
action: 'detectLabels',
iamResources: ['*'],
additionalIamStatements: [
new iam.PolicyStatement({
actions: ['s3:getObject'],
resources: ['arn:aws:s3:::amzn-s3-demo-bucket/*'],
}),
],
});
python
detect_labels = tasks.CallAwsService(self, "DetectLabels",
service="rekognition",
action="detectLabels",
iam_resources=["*"],
additional_iam_statements=[
iam.PolicyStatement(
actions=["s3:getObject"],
resources=["arn:aws:s3:::amzn-s3-demo-bucket/*"]
)
]
)
java
CallAwsService detectLabels = CallAwsService.Builder.create(this, "DetectLabels")
.service("rekognition")
.action("detectLabels")
.iamResources(List.of("*"))
.additionalIamStatements(List.of(
PolicyStatement.Builder.create()
.actions(List.of("s3:getObject"))
.resources(List.of("arn:aws:s3:::amzn-s3-demo-bucket/*"))
.build()))
.build();
csharp
var detectLabels = new CallAwsService(this, "DetectLabels", new CallAwsServiceProps {
Service = "rekognition",
Action = "detectLabels",
IamResources = new [] { "*" },
AdditionalIamStatements = new [] {
new PolicyStatement(new PolicyStatementProps {
Actions = new [] { "s3:getObject" },
Resources = new [] { "arn:aws:s3:::amzn-s3-demo-bucket/*" }
}) }
});
go
detectLabels := tasks.NewCallAwsService(this, jsii.String("DetectLabels"), &CallAwsServiceProps{
Service: jsii.String("rekognition"),
Action: jsii.String("detectLabels"),
IamResources: []*string{
jsii.String("*"),
},
AdditionalIamStatements: []PolicyStatement{
iam.NewPolicyStatement(&PolicyStatementProps{
Actions: []*string{
jsii.String("s3:getObject"),
},
Resources: []*string{
jsii.String("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.
ts
declare const myBucket: s3.Bucket;
const getObject = new tasks.CallAwsServiceCrossRegion(this, 'GetObject', {
region: 'ap-northeast-1',
service: 's3',
action: 'getObject',
parameters: {
Bucket: myBucket.bucketName,
Key: sfn.JsonPath.stringAt('$.key')
},
iamResources: [myBucket.arnForObjects('*')],
});
python
# my_bucket: s3.Bucket
get_object = tasks.CallAwsServiceCrossRegion(self, "GetObject",
region="ap-northeast-1",
service="s3",
action="getObject",
parameters={
"Bucket": my_bucket.bucket_name,
"Key": sfn.JsonPath.string_at("$.key")
},
iam_resources=[my_bucket.arn_for_objects("*")]
)
java
Bucket myBucket;
CallAwsServiceCrossRegion getObject = CallAwsServiceCrossRegion.Builder.create(this, "GetObject")
.region("ap-northeast-1")
.service("s3")
.action("getObject")
.parameters(Map.of(
"Bucket", myBucket.getBucketName(),
"Key", JsonPath.stringAt("$.key")))
.iamResources(List.of(myBucket.arnForObjects("*")))
.build();
csharp
Bucket myBucket;
var getObject = new CallAwsServiceCrossRegion(this, "GetObject", new CallAwsServiceCrossRegionProps {
Region = "ap-northeast-1",
Service = "s3",
Action = "getObject",
Parameters = new Dictionary<string, object> {
{ "Bucket", myBucket.BucketName },
{ "Key", JsonPath.StringAt("$.key") }
},
IamResources = new [] { myBucket.ArnForObjects("*") }
});
go
var myBucket Bucket
getObject := tasks.NewCallAwsServiceCrossRegion(this, jsii.String("GetObject"), &CallAwsServiceCrossRegionProps{
Region: jsii.String("ap-northeast-1"),
Service: jsii.String("s3"),
Action: jsii.String("getObject"),
Parameters: map[string]interface{}{
"Bucket": myBucket.bucketName,
"Key": sfn.JsonPath_stringAt(jsii.String("$.key")),
},
IamResources: []*string{
myBucket.ArnForObjects(jsii.String("*")),
},
})
Other properties such as additionalIamStatements 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.
ts
const startQueryExecutionJob = new tasks.AthenaStartQueryExecution(this, 'Start Athena Query', {
queryString: sfn.JsonPath.stringAt('$.queryString'),
queryExecutionContext: {
databaseName: 'mydatabase',
},
resultConfiguration: {
encryptionConfiguration: {
encryptionOption: tasks.EncryptionOption.S3_MANAGED,
},
outputLocation: {
bucketName: 'amzn-s3-demo-bucket',
objectKey: 'folder',
},
},
executionParameters: ['param1', 'param2'],
});
python
start_query_execution_job = tasks.AthenaStartQueryExecution(self, "Start Athena Query",
query_string=sfn.JsonPath.string_at("$.queryString"),
query_execution_context=tasks.QueryExecutionContext(
database_name="mydatabase"
),
result_configuration=tasks.ResultConfiguration(
encryption_configuration=tasks.EncryptionConfiguration(
encryption_option=tasks.EncryptionOption.S3_MANAGED
),
output_location=s3.Location(
bucket_name="amzn-s3-demo-bucket",
object_key="folder"
)
),
execution_parameters=["param1", "param2"]
)
java
AthenaStartQueryExecution startQueryExecutionJob = AthenaStartQueryExecution.Builder.create(this, "Start Athena Query")
.queryString(JsonPath.stringAt("$.queryString"))
.queryExecutionContext(QueryExecutionContext.builder()
.databaseName("mydatabase")
.build())
.resultConfiguration(ResultConfiguration.builder()
.encryptionConfiguration(EncryptionConfiguration.builder()
.encryptionOption(EncryptionOption.S3_MANAGED)
.build())
.outputLocation(Location.builder()
.bucketName("amzn-s3-demo-bucket")
.objectKey("folder")
.build())
.build())
.executionParameters(List.of("param1", "param2"))
.build();
csharp
var startQueryExecutionJob = new AthenaStartQueryExecution(this, "Start Athena Query", new AthenaStartQueryExecutionProps {
QueryString = JsonPath.StringAt("$.queryString"),
QueryExecutionContext = new QueryExecutionContext {
DatabaseName = "mydatabase"
},
ResultConfiguration = new ResultConfiguration {
EncryptionConfiguration = new EncryptionConfiguration {
EncryptionOption = EncryptionOption.S3_MANAGED
},
OutputLocation = new Location {
BucketName = "amzn-s3-demo-bucket",
ObjectKey = "folder"
}
},
ExecutionParameters = new [] { "param1", "param2" }
});
go
startQueryExecutionJob := tasks.NewAthenaStartQueryExecution(this, jsii.String("Start Athena Query"), &AthenaStartQueryExecutionProps{
QueryString: sfn.JsonPath_StringAt(jsii.String("$.queryString")),
QueryExecutionContext: &QueryExecutionContext{
DatabaseName: jsii.String("mydatabase"),
},
ResultConfiguration: &ResultConfiguration{
EncryptionConfiguration: &EncryptionConfiguration{
EncryptionOption: tasks.EncryptionOption_S3_MANAGED,
},
OutputLocation: &Location{
BucketName: jsii.String("amzn-s3-demo-bucket"),
ObjectKey: jsii.String("folder"),
},
},
ExecutionParameters: []*string{
jsii.String("param1"),
jsii.String("param2"),
},
})
You can reuse the query results by setting the resultReuseConfigurationMaxAge property.
ts
const startQueryExecutionJob = new tasks.AthenaStartQueryExecution(this, 'Start Athena Query', {
queryString: sfn.JsonPath.stringAt('$.queryString'),
queryExecutionContext: {
databaseName: 'mydatabase',
},
resultConfiguration: {
encryptionConfiguration: {
encryptionOption: tasks.EncryptionOption.S3_MANAGED,
},
outputLocation: {
bucketName: 'query-results-bucket',
objectKey: 'folder',
},
},
executionParameters: ['param1', 'param2'],
resultReuseConfigurationMaxAge: Duration.minutes(100),
});
python
start_query_execution_job = tasks.AthenaStartQueryExecution(self, "Start Athena Query",
query_string=sfn.JsonPath.string_at("$.queryString"),
query_execution_context=tasks.QueryExecutionContext(
database_name="mydatabase"
),
result_configuration=tasks.ResultConfiguration(
encryption_configuration=tasks.EncryptionConfiguration(
encryption_option=tasks.EncryptionOption.S3_MANAGED
),
output_location=s3.Location(
bucket_name="query-results-bucket",
object_key="folder"
)
),
execution_parameters=["param1", "param2"],
result_reuse_configuration_max_age=Duration.minutes(100)
)
java
AthenaStartQueryExecution startQueryExecutionJob = AthenaStartQueryExecution.Builder.create(this, "Start Athena Query")
.queryString(JsonPath.stringAt("$.queryString"))
.queryExecutionContext(QueryExecutionContext.builder()
.databaseName("mydatabase")
.build())
.resultConfiguration(ResultConfiguration.builder()
.encryptionConfiguration(EncryptionConfiguration.builder()
.encryptionOption(EncryptionOption.S3_MANAGED)
.build())
.outputLocation(Location.builder()
.bucketName("query-results-bucket")
.objectKey("folder")
.build())
.build())
.executionParameters(List.of("param1", "param2"))
.resultReuseConfigurationMaxAge(Duration.minutes(100))
.build();
csharp
var startQueryExecutionJob = new AthenaStartQueryExecution(this, "Start Athena Query", new AthenaStartQueryExecutionProps {
QueryString = JsonPath.StringAt("$.queryString"),
QueryExecutionContext = new QueryExecutionContext {
DatabaseName = "mydatabase"
},
ResultConfiguration = new ResultConfiguration {
EncryptionConfiguration = new EncryptionConfiguration {
EncryptionOption = EncryptionOption.S3_MANAGED
},
OutputLocation = new Location {
BucketName = "query-results-bucket",
ObjectKey = "folder"
}
},
ExecutionParameters = new [] { "param1", "param2" },
ResultReuseConfigurationMaxAge = Duration.Minutes(100)
});
go
startQueryExecutionJob := tasks.NewAthenaStartQueryExecution(this, jsii.String("Start Athena Query"), &AthenaStartQueryExecutionProps{
QueryString: sfn.JsonPath_StringAt(jsii.String("$.queryString")),
QueryExecutionContext: &QueryExecutionContext{
DatabaseName: jsii.String("mydatabase"),
},
ResultConfiguration: &ResultConfiguration{
EncryptionConfiguration: &EncryptionConfiguration{
EncryptionOption: tasks.EncryptionOption_S3_MANAGED,
},
OutputLocation: &Location{
BucketName: jsii.String("query-results-bucket"),
ObjectKey: jsii.String("folder"),
},
},
ExecutionParameters: []*string{
jsii.String("param1"),
jsii.String("param2"),
},
ResultReuseConfigurationMaxAge: awscdk.Duration_Minutes(jsii.Number(100)),
})
GetQueryExecution
The GetQueryExecution API gets information about a single execution of a query.
ts
const getQueryExecutionJob = new tasks.AthenaGetQueryExecution(this, 'Get Query Execution', {
queryExecutionId: sfn.JsonPath.stringAt('$.QueryExecutionId'),
});
python
get_query_execution_job = tasks.AthenaGetQueryExecution(self, "Get Query Execution",
query_execution_id=sfn.JsonPath.string_at("$.QueryExecutionId")
)
java
AthenaGetQueryExecution getQueryExecutionJob = AthenaGetQueryExecution.Builder.create(this, "Get Query Execution")
.queryExecutionId(JsonPath.stringAt("$.QueryExecutionId"))
.build();
csharp
var getQueryExecutionJob = new AthenaGetQueryExecution(this, "Get Query Execution", new AthenaGetQueryExecutionProps {
QueryExecutionId = JsonPath.StringAt("$.QueryExecutionId")
});
go
getQueryExecutionJob := tasks.NewAthenaGetQueryExecution(this, jsii.String("Get Query Execution"), &AthenaGetQueryExecutionProps{
QueryExecutionId: sfn.JsonPath_StringAt(jsii.String("$.QueryExecutionId")),
})
GetQueryResults
The GetQueryResults API that streams the results of a single query execution specified by QueryExecutionId from S3.
ts
const getQueryResultsJob = new tasks.AthenaGetQueryResults(this, 'Get Query Results', {
queryExecutionId: sfn.JsonPath.stringAt('$.QueryExecutionId'),
});
python
get_query_results_job = tasks.AthenaGetQueryResults(self, "Get Query Results",
query_execution_id=sfn.JsonPath.string_at("$.QueryExecutionId")
)
java
AthenaGetQueryResults getQueryResultsJob = AthenaGetQueryResults.Builder.create(this, "Get Query Results")
.queryExecutionId(JsonPath.stringAt("$.QueryExecutionId"))
.build();
csharp
var getQueryResultsJob = new AthenaGetQueryResults(this, "Get Query Results", new AthenaGetQueryResultsProps {
QueryExecutionId = JsonPath.StringAt("$.QueryExecutionId")
});
go
getQueryResultsJob := tasks.NewAthenaGetQueryResults(this, jsii.String("Get Query Results"), &AthenaGetQueryResultsProps{
QueryExecutionId: sfn.JsonPath_StringAt(jsii.String("$.QueryExecutionId")),
})
StopQueryExecution
The StopQueryExecution API that stops a query execution.
ts
const stopQueryExecutionJob = new tasks.AthenaStopQueryExecution(this, 'Stop Query Execution', {
queryExecutionId: sfn.JsonPath.stringAt('$.QueryExecutionId'),
});
python
stop_query_execution_job = tasks.AthenaStopQueryExecution(self, "Stop Query Execution",
query_execution_id=sfn.JsonPath.string_at("$.QueryExecutionId")
)
java
AthenaStopQueryExecution stopQueryExecutionJob = AthenaStopQueryExecution.Builder.create(this, "Stop Query Execution")
.queryExecutionId(JsonPath.stringAt("$.QueryExecutionId"))
.build();
csharp
var stopQueryExecutionJob = new AthenaStopQueryExecution(this, "Stop Query Execution", new AthenaStopQueryExecutionProps {
QueryExecutionId = JsonPath.StringAt("$.QueryExecutionId")
});
go
stopQueryExecutionJob := tasks.NewAthenaStopQueryExecution(this, jsii.String("Stop Query Execution"), &AthenaStopQueryExecutionProps{
QueryExecutionId: sfn.JsonPath_StringAt(jsii.String("$.QueryExecutionId")),
})
Batch
Step Functions supports Batch through the service integration pattern.
SubmitJob
The SubmitJob API submits an AWS Batch job from a job definition.
ts
import * as batch from 'aws-cdk-lib/aws-batch';
declare const batchJobDefinition: batch.EcsJobDefinition;
declare const batchQueue: batch.JobQueue;
const task = new tasks.BatchSubmitJob(this, 'Submit Job', {
jobDefinitionArn: batchJobDefinition.jobDefinitionArn,
jobName: 'MyJob',
jobQueueArn: batchQueue.jobQueueArn,
});
python
import aws_cdk.aws_batch as batch
# batch_job_definition: batch.EcsJobDefinition
# batch_queue: batch.JobQueue
task = tasks.BatchSubmitJob(self, "Submit Job",
job_definition_arn=batch_job_definition.job_definition_arn,
job_name="MyJob",
job_queue_arn=batch_queue.job_queue_arn
)
java
import software.amazon.awscdk.services.batch.*;
EcsJobDefinition batchJobDefinition;
JobQueue batchQueue;
BatchSubmitJob task = BatchSubmitJob.Builder.create(this, "Submit Job")
.jobDefinitionArn(batchJobDefinition.getJobDefinitionArn())
.jobName("MyJob")
.jobQueueArn(batchQueue.getJobQueueArn())
.build();
csharp
using Amazon.CDK.AWS.Batch;
EcsJobDefinition batchJobDefinition;
JobQueue batchQueue;
var task = new BatchSubmitJob(this, "Submit Job", new BatchSubmitJobProps {
JobDefinitionArn = batchJobDefinition.JobDefinitionArn,
JobName = "MyJob",
JobQueueArn = batchQueue.JobQueueArn
});
go
import batch "github.com/aws/aws-cdk-go/awscdk"
var batchJobDefinition EcsJobDefinition
var batchQueue JobQueue
task := tasks.NewBatchSubmitJob(this, jsii.String("Submit Job"), &BatchSubmitJobProps{
JobDefinitionArn: batchJobDefinition.jobDefinitionArn,
JobName: jsii.String("MyJob"),
JobQueueArn: batchQueue.jobQueueArn,
})
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.
ts
import * as bedrock from 'aws-cdk-lib/aws-bedrock';
const model = bedrock.FoundationModel.fromFoundationModelId(
this,
'Model',
bedrock.FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1,
);
const task = new tasks.BedrockInvokeModel(this, 'Prompt Model', {
model,
body: sfn.TaskInput.fromObject(
{
inputText: 'Generate a list of five first names.',
textGenerationConfig: {
maxTokenCount: 100,
temperature: 1,
},
},
),
resultSelector: {
names: sfn.JsonPath.stringAt('$.Body.results[0].outputText'),
},
});
python
import aws_cdk.aws_bedrock as bedrock
model = bedrock.FoundationModel.from_foundation_model_id(self, "Model", bedrock.FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1)
task = tasks.BedrockInvokeModel(self, "Prompt Model",
model=model,
body=sfn.TaskInput.from_object({
"input_text": "Generate a list of five first names.",
"text_generation_config": {
"max_token_count": 100,
"temperature": 1
}
}),
result_selector={
"names": sfn.JsonPath.string_at("$.Body.results[0].outputText")
}
)
java
import software.amazon.awscdk.services.bedrock.*;
FoundationModel model = FoundationModel.fromFoundationModelId(this, "Model", FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1);
BedrockInvokeModel task = BedrockInvokeModel.Builder.create(this, "Prompt Model")
.model(model)
.body(TaskInput.fromObject(Map.of(
"inputText", "Generate a list of five first names.",
"textGenerationConfig", Map.of(
"maxTokenCount", 100,
"temperature", 1))))
.resultSelector(Map.of(
"names", JsonPath.stringAt("$.Body.results[0].outputText")))
.build();
csharp
using Amazon.CDK.AWS.Bedrock;
var model = FoundationModel.FromFoundationModelId(this, "Model", FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1);
var task = new BedrockInvokeModel(this, "Prompt Model", new BedrockInvokeModelProps {
Model = model,
Body = TaskInput.FromObject(new Dictionary<string, object> {
{ "inputText", "Generate a list of five first names." },
{ "textGenerationConfig", new Dictionary<string, int> {
{ "maxTokenCount", 100 },
{ "temperature", 1 }
} }
}),
ResultSelector = new Dictionary<string, object> {
{ "names", JsonPath.StringAt("$.Body.results[0].outputText") }
}
});
go
import "github.com/aws/aws-cdk-go/awscdk"
model := bedrock.FoundationModel_FromFoundationModelId(this, jsii.String("Model"), bedrock.FoundationModelIdentifier_AMAZON_TITAN_TEXT_G1_EXPRESS_V1())
task := tasks.NewBedrockInvokeModel(this, jsii.String("Prompt Model"), &BedrockInvokeModelProps{
Model: Model,
Body: sfn.TaskInput_FromObject(map[string]interface{}{
"inputText": jsii.String("Generate a list of five first names."),
"textGenerationConfig": map[string]*f64{
"maxTokenCount": jsii.Number(100),
"temperature": jsii.Number(1),
},
}),
ResultSelector: map[string]interface{}{
"names": sfn.JsonPath_stringAt(jsii.String("$.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 s3InputUri and s3OutputUri 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.
ts
import * as bedrock from 'aws-cdk-lib/aws-bedrock';
const model = bedrock.FoundationModel.fromFoundationModelId(
this,
'Model',
bedrock.FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1,
);
const task = new tasks.BedrockInvokeModel(this, 'Prompt Model', {
model,
input : { s3InputUri: sfn.JsonPath.stringAt('$.prompt') },
output: { s3OutputUri: sfn.JsonPath.stringAt('$.prompt') },
});
python
import aws_cdk.aws_bedrock as bedrock
model = bedrock.FoundationModel.from_foundation_model_id(self, "Model", bedrock.FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1)
task = tasks.BedrockInvokeModel(self, "Prompt Model",
model=model,
input=tasks.BedrockInvokeModelInputProps(s3_input_uri=sfn.JsonPath.string_at("$.prompt")),
output=tasks.BedrockInvokeModelOutputProps(s3_output_uri=sfn.JsonPath.string_at("$.prompt"))
)
java
import software.amazon.awscdk.services.bedrock.*;
FoundationModel model = FoundationModel.fromFoundationModelId(this, "Model", FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1);
BedrockInvokeModel task = BedrockInvokeModel.Builder.create(this, "Prompt Model")
.model(model)
.input(BedrockInvokeModelInputProps.builder().s3InputUri(JsonPath.stringAt("$.prompt")).build())
.output(BedrockInvokeModelOutputProps.builder().s3OutputUri(JsonPath.stringAt("$.prompt")).build())
.build();
csharp
using Amazon.CDK.AWS.Bedrock;
var model = FoundationModel.FromFoundationModelId(this, "Model", FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1);
var task = new BedrockInvokeModel(this, "Prompt Model", new BedrockInvokeModelProps {
Model = model,
Input = new BedrockInvokeModelInputProps { S3InputUri = JsonPath.StringAt("$.prompt") },
Output = new BedrockInvokeModelOutputProps { S3OutputUri = JsonPath.StringAt("$.prompt") }
});
go
import "github.com/aws/aws-cdk-go/awscdk"
model := bedrock.FoundationModel_FromFoundationModelId(this, jsii.String("Model"), bedrock.FoundationModelIdentifier_AMAZON_TITAN_TEXT_G1_EXPRESS_V1())
task := tasks.NewBedrockInvokeModel(this, jsii.String("Prompt Model"), &BedrockInvokeModelProps{
Model: Model,
Input: &BedrockInvokeModelInputProps{
S3InputUri: sfn.JsonPath_StringAt(jsii.String("$.prompt")),
},
Output: &BedrockInvokeModelOutputProps{
S3OutputUri: sfn.JsonPath_*StringAt(jsii.String("$.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(s3InputUri and s3OutputUri), 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.
ts
import * as bedrock from 'aws-cdk-lib/aws-bedrock';
const model = bedrock.FoundationModel.fromFoundationModelId(
this,
'Model',
bedrock.FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1,
);
const task = new tasks.BedrockInvokeModel(this, 'Prompt Model', {
model,
inputPath: sfn.JsonPath.stringAt('$.prompt'),
outputPath: sfn.JsonPath.stringAt('$.prompt'),
});
python
import aws_cdk.aws_bedrock as bedrock
model = bedrock.FoundationModel.from_foundation_model_id(self, "Model", bedrock.FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1)
task = tasks.BedrockInvokeModel(self, "Prompt Model",
model=model,
input_path=sfn.JsonPath.string_at("$.prompt"),
output_path=sfn.JsonPath.string_at("$.prompt")
)
java
import software.amazon.awscdk.services.bedrock.*;
FoundationModel model = FoundationModel.fromFoundationModelId(this, "Model", FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1);
BedrockInvokeModel task = BedrockInvokeModel.Builder.create(this, "Prompt Model")
.model(model)
.inputPath(JsonPath.stringAt("$.prompt"))
.outputPath(JsonPath.stringAt("$.prompt"))
.build();
csharp
using Amazon.CDK.AWS.Bedrock;
var model = FoundationModel.FromFoundationModelId(this, "Model", FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1);
var task = new BedrockInvokeModel(this, "Prompt Model", new BedrockInvokeModelProps {
Model = model,
InputPath = JsonPath.StringAt("$.prompt"),
OutputPath = JsonPath.StringAt("$.prompt")
});
go
import "github.com/aws/aws-cdk-go/awscdk"
model := bedrock.FoundationModel_FromFoundationModelId(this, jsii.String("Model"), bedrock.FoundationModelIdentifier_AMAZON_TITAN_TEXT_G1_EXPRESS_V1())
task := tasks.NewBedrockInvokeModel(this, jsii.String("Prompt Model"), &BedrockInvokeModelProps{
Model: Model,
InputPath: sfn.JsonPath_StringAt(jsii.String("$.prompt")),
OutputPath: sfn.JsonPath_*StringAt(jsii.String("$.prompt")),
})
You can apply a guardrail to the invocation by setting guardrail.
ts
import * as bedrock from 'aws-cdk-lib/aws-bedrock';
const model = bedrock.FoundationModel.fromFoundationModelId(
this,
'Model',
bedrock.FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1,
);
const task = new tasks.BedrockInvokeModel(this, 'Prompt Model with guardrail', {
model,
body: sfn.TaskInput.fromObject(
{
inputText: 'Generate a list of five first names.',
textGenerationConfig: {
maxTokenCount: 100,
temperature: 1,
},
},
),
guardrail: tasks.Guardrail.enable('guardrailId', 1),
resultSelector: {
names: sfn.JsonPath.stringAt('$.Body.results[0].outputText'),
},
});
python
import aws_cdk.aws_bedrock as bedrock
model = bedrock.FoundationModel.from_foundation_model_id(self, "Model", bedrock.FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1)
task = tasks.BedrockInvokeModel(self, "Prompt Model with guardrail",
model=model,
body=sfn.TaskInput.from_object({
"input_text": "Generate a list of five first names.",
"text_generation_config": {
"max_token_count": 100,
"temperature": 1
}
}),
guardrail=tasks.Guardrail.enable("guardrailId", 1),
result_selector={
"names": sfn.JsonPath.string_at("$.Body.results[0].outputText")
}
)
java
import software.amazon.awscdk.services.bedrock.*;
FoundationModel model = FoundationModel.fromFoundationModelId(this, "Model", FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1);
BedrockInvokeModel task = BedrockInvokeModel.Builder.create(this, "Prompt Model with guardrail")
.model(model)
.body(TaskInput.fromObject(Map.of(
"inputText", "Generate a list of five first names.",
"textGenerationConfig", Map.of(
"maxTokenCount", 100,
"temperature", 1))))
.guardrail(Guardrail.enable("guardrailId", 1))
.resultSelector(Map.of(
"names", JsonPath.stringAt("$.Body.results[0].outputText")))
.build();
csharp
using Amazon.CDK.AWS.Bedrock;
var model = FoundationModel.FromFoundationModelId(this, "Model", FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1);
var task = new BedrockInvokeModel(this, "Prompt Model with guardrail", new BedrockInvokeModelProps {
Model = model,
Body = TaskInput.FromObject(new Dictionary<string, object> {
{ "inputText", "Generate a list of five first names." },
{ "textGenerationConfig", new Dictionary<string, int> {
{ "maxTokenCount", 100 },
{ "temperature", 1 }
} }
}),
Guardrail = Guardrail.Enable("guardrailId", 1),
ResultSelector = new Dictionary<string, object> {
{ "names", JsonPath.StringAt("$.Body.results[0].outputText") }
}
});
go
import "github.com/aws/aws-cdk-go/awscdk"
model := bedrock.FoundationModel_FromFoundationModelId(this, jsii.String("Model"), bedrock.FoundationModelIdentifier_AMAZON_TITAN_TEXT_G1_EXPRESS_V1())
task := tasks.NewBedrockInvokeModel(this, jsii.String("Prompt Model with guardrail"), &BedrockInvokeModelProps{
Model: Model,
Body: sfn.TaskInput_FromObject(map[string]interface{}{
"inputText": jsii.String("Generate a list of five first names."),
"textGenerationConfig": map[string]*f64{
"maxTokenCount": jsii.Number(100),
"temperature": jsii.Number(1),
},
}),
Guardrail: tasks.Guardrail_Enable(jsii.String("guardrailId"), jsii.Number(1)),
ResultSelector: map[string]interface{}{
"names": sfn.JsonPath_stringAt(jsii.String("$.Body.results[0].outputText")),
},
})
createModelCustomizationJob
The CreateModelCustomizationJob API creates a fine-tuning job to customize a base model.
ts
import * as bedrock from 'aws-cdk-lib/aws-bedrock';
import * as kms from 'aws-cdk-lib/aws-kms';
declare const outputBucket: s3.IBucket;
declare const trainingBucket: s3.IBucket;
declare const validationBucket: s3.IBucket;
declare const kmsKey: kms.IKey;
declare const vpc: ec2.IVpc;
const model = bedrock.FoundationModel.fromFoundationModelId(
this,
'Model',
bedrock.FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1,
);
const task = new tasks.BedrockCreateModelCustomizationJob(this, 'CreateModelCustomizationJob', {
baseModel: model,
clientRequestToken: 'MyToken',
customizationType: tasks.CustomizationType.FINE_TUNING,
customModelKmsKey: kmsKey,
customModelName: 'MyCustomModel', // required
customModelTags: [{ key: 'key1', value: 'value1' }],
hyperParameters: {
batchSize: '10',
},
jobName: 'MyCustomizationJob', // required
jobTags: [{ key: 'key2', value: 'value2' }],
outputData: {
bucket: outputBucket, // required
path: 'output-data/',
},
trainingData: {
bucket: trainingBucket,
path: 'training-data/data.json',
}, // required
// If you don't provide validation data, you have to specify `Evaluation percentage` hyperparameter.
validationData: [
{
bucket: validationBucket,
path: 'validation-data/data.json',
},
],
vpcConfig: {
securityGroups: [new ec2.SecurityGroup(this, 'SecurityGroup', { vpc })],
subnets: vpc.privateSubnets,
},
});
python
import aws_cdk.aws_bedrock as bedrock
import aws_cdk.aws_kms as kms
# output_bucket: s3.IBucket
# training_bucket: s3.IBucket
# validation_bucket: s3.IBucket
# kms_key: kms.IKey
# vpc: ec2.IVpc
model = bedrock.FoundationModel.from_foundation_model_id(self, "Model", bedrock.FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1)
task = tasks.BedrockCreateModelCustomizationJob(self, "CreateModelCustomizationJob",
base_model=model,
client_request_token="MyToken",
customization_type=tasks.CustomizationType.FINE_TUNING,
custom_model_kms_key=kms_key,
custom_model_name="MyCustomModel", # required
custom_model_tags=[tasks.CustomModelTag(key="key1", value="value1")],
hyper_parameters={
"batch_size": "10"
},
job_name="MyCustomizationJob", # required
job_tags=[tasks.CustomModelTag(key="key2", value="value2")],
output_data=tasks.OutputBucketConfiguration(
bucket=output_bucket, # required
path="output-data/"
),
training_data=tasks.TrainingBucketConfiguration(
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=[tasks.ValidationBucketConfiguration(
bucket=validation_bucket,
path="validation-data/data.json"
)
],
vpc_config={
"security_groups": [ec2.SecurityGroup(self, "SecurityGroup", vpc=vpc)],
"subnets": vpc.private_subnets
}
)
java
import software.amazon.awscdk.services.bedrock.*;
import software.amazon.awscdk.services.kms.*;
IBucket outputBucket;
IBucket trainingBucket;
IBucket validationBucket;
IKey kmsKey;
IVpc vpc;
FoundationModel model = FoundationModel.fromFoundationModelId(this, "Model", FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1);
BedrockCreateModelCustomizationJob task = BedrockCreateModelCustomizationJob.Builder.create(this, "CreateModelCustomizationJob")
.baseModel(model)
.clientRequestToken("MyToken")
.customizationType(CustomizationType.FINE_TUNING)
.customModelKmsKey(kmsKey)
.customModelName("MyCustomModel") // required
.customModelTags(List.of(CustomModelTag.builder().key("key1").value("value1").build()))
.hyperParameters(Map.of(
"batchSize", "10"))
.jobName("MyCustomizationJob") // required
.jobTags(List.of(CustomModelTag.builder().key("key2").value("value2").build()))
.outputData(OutputBucketConfiguration.builder()
.bucket(outputBucket) // required
.path("output-data/")
.build())
.trainingData(TrainingBucketConfiguration.builder()
.bucket(trainingBucket)
.path("training-data/data.json")
.build()) // required
// If you don't provide validation data, you have to specify `Evaluation percentage` hyperparameter.
.validationData(List.of(ValidationBucketConfiguration.builder()
.bucket(validationBucket)
.path("validation-data/data.json")
.build()))
.vpcConfig(Map.of(
"securityGroups", List.of(SecurityGroup.Builder.create(this, "SecurityGroup").vpc(vpc).build()),
"subnets", vpc.getPrivateSubnets()))
.build();
csharp
using Amazon.CDK.AWS.Bedrock;
using Amazon.CDK.AWS.KMS;
IBucket outputBucket;
IBucket trainingBucket;
IBucket validationBucket;
IKey kmsKey;
IVpc vpc;
var model = FoundationModel.FromFoundationModelId(this, "Model", FoundationModelIdentifier.AMAZON_TITAN_TEXT_G1_EXPRESS_V1);
var task = new BedrockCreateModelCustomizationJob(this, "CreateModelCustomizationJob", new BedrockCreateModelCustomizationJobProps {
BaseModel = model,
ClientRequestToken = "MyToken",
CustomizationType = CustomizationType.FINE_TUNING,
CustomModelKmsKey = kmsKey,
CustomModelName = "MyCustomModel", // required
CustomModelTags = new [] { new CustomModelTag { Key = "key1", Value = "value1" } },
HyperParameters = new Dictionary<string, string> {
{ "batchSize", "10" }
},
JobName = "MyCustomizationJob", // required
JobTags = new [] { new CustomModelTag { Key = "key2", Value = "value2" } },
OutputData = new OutputBucketConfiguration {
Bucket = outputBucket, // required
Path = "output-data/"
},
TrainingData = new TrainingBucketConfiguration {
Bucket = trainingBucket,
Path = "training-data/data.json"
}, // required
// If you don't provide validation data, you have to specify `Evaluation percentage` hyperparameter.
ValidationData = new [] { new ValidationBucketConfiguration {
Bucket = validationBucket,
Path = "validation-data/data.json"
} },
VpcConfig = new Dictionary<string, ISecurityGroup[]> {
{ "securityGroups", new [] { new SecurityGroup(this, "SecurityGroup", new SecurityGroupProps { Vpc = vpc }) } },
{ "subnets", vpc.PrivateSubnets }
}
});
go
import "github.com/aws/aws-cdk-go/awscdk"
import kms "github.com/aws/aws-cdk-go/awscdk"
var outputBucket IBucket
var trainingBucket IBucket
var validationBucket IBucket
var kmsKey IKey
var vpc IVpc
model := bedrock.FoundationModel_FromFoundationModelId(this, jsii.String("Model"), bedrock.FoundationModelIdentifier_AMAZON_TITAN_TEXT_G1_EXPRESS_V1())
task := tasks.NewBedrockCreateModelCustomizationJob(this, jsii.String("CreateModelCustomizationJob"), &BedrockCreateModelCustomizationJobProps{
BaseModel: model,
ClientRequestToken: jsii.String("MyToken"),
CustomizationType: tasks.CustomizationType_FINE_TUNING,
CustomModelKmsKey: kmsKey,
CustomModelName: jsii.String("MyCustomModel"),
// required
CustomModelTags: []CustomModelTag{
&CustomModelTag{
Key: jsii.String("key1"),
Value: jsii.String("value1"),
},
},
HyperParameters: map[string]*string{
"batchSize": jsii.String("10"),
},
JobName: jsii.String("MyCustomizationJob"),
// required
JobTags: []CustomModelTag{
&CustomModelTag{
Key: jsii.String("key2"),
Value: jsii.String("value2"),
},
},
OutputData: &OutputBucketConfiguration{
Bucket: outputBucket,
// required
Path: jsii.String("output-data/"),
},
TrainingData: &TrainingBucketConfiguration{
Bucket: trainingBucket,
Path: jsii.String("training-data/data.json"),
},
// required
// If you don't provide validation data, you have to specify `Evaluation percentage` hyperparameter.
ValidationData: []ValidationBucketConfiguration{
&ValidationBucketConfiguration{
Bucket: validationBucket,
Path: jsii.String("validation-data/data.json"),
},
},
VpcConfig: map[string][]ISecurityGroup{
"securityGroups": []ISecurityGroup{
ec2.NewSecurityGroup(this, jsii.String("SecurityGroup"), &SecurityGroupProps{
"vpc": vpc,
}),
},
"subnets": vpc.privateSubnets,
},
})
CodeBuild
Step Functions supports CodeBuild through the service integration pattern.
StartBuild
StartBuild starts a CodeBuild Project by Project Name.
ts
import * as codebuild from 'aws-cdk-lib/aws-codebuild';
const codebuildProject = new codebuild.Project(this, 'Project', {
projectName: 'MyTestProject',
buildSpec: codebuild.BuildSpec.fromObject({
version: '0.2',
phases: {
build: {
commands: [
'echo "Hello, CodeBuild!"',
],
},
},
}),
});
const task = new tasks.CodeBuildStartBuild(this, 'Task', {
project: codebuildProject,
integrationPattern: sfn.IntegrationPattern.RUN_JOB,
environmentVariablesOverride: {
ZONE: {
type: codebuild.BuildEnvironmentVariableType.PLAINTEXT,
value: sfn.JsonPath.stringAt('$.envVariables.zone'),
},
},
});
python
import aws_cdk.aws_codebuild as codebuild
codebuild_project = codebuild.Project(self, "Project",
project_name="MyTestProject",
build_spec=codebuild.BuildSpec.from_object({
"version": "0.2",
"phases": {
"build": {
"commands": ["echo \"Hello, CodeBuild!\""
]
}
}
})
)
task = tasks.CodeBuildStartBuild(self, "Task",
project=codebuild_project,
integration_pattern=sfn.IntegrationPattern.RUN_JOB,
environment_variables_override={
"ZONE": codebuild.BuildEnvironmentVariable(
type=codebuild.BuildEnvironmentVariableType.PLAINTEXT,
value=sfn.JsonPath.string_at("$.envVariables.zone")
)
}
)
java
import software.amazon.awscdk.services.codebuild.*;
Project codebuildProject = Project.Builder.create(this, "Project")
.projectName("MyTestProject")
.buildSpec(BuildSpec.fromObject(Map.of(
"version", "0.2",
"phases", Map.of(
"build", Map.of(
"commands", List.of("echo \"Hello, CodeBuild!\""))))))
.build();
CodeBuildStartBuild task = CodeBuildStartBuild.Builder.create(this, "Task")
.project(codebuildProject)
.integrationPattern(IntegrationPattern.RUN_JOB)
.environmentVariablesOverride(Map.of(
"ZONE", BuildEnvironmentVariable.builder()
.type(BuildEnvironmentVariableType.PLAINTEXT)
.value(JsonPath.stringAt("$.envVariables.zone"))
.build()))
.build();
csharp
using Amazon.CDK.AWS.CodeBuild;
var codebuildProject = new Project(this, "Project", new ProjectProps {
ProjectName = "MyTestProject",
BuildSpec = BuildSpec.FromObject(new Dictionary<string, object> {
{ "version", "0.2" },
{ "phases", new Dictionary<string, IDictionary<string, string[]>> {
{ "build", new Struct {
Commands = new [] { "echo \"Hello, CodeBuild!\"" }
} }
} }
})
});
var task = new CodeBuildStartBuild(this, "Task", new CodeBuildStartBuildProps {
Project = codebuildProject,
IntegrationPattern = IntegrationPattern.RUN_JOB,
EnvironmentVariablesOverride = new Dictionary<string, BuildEnvironmentVariable> {
{ "ZONE", new BuildEnvironmentVariable {
Type = BuildEnvironmentVariableType.PLAINTEXT,
Value = JsonPath.StringAt("$.envVariables.zone")
} }
}
});
go
import "github.com/aws/aws-cdk-go/awscdk"
codebuildProject := codebuild.NewProject(this, jsii.String("Project"), &ProjectProps{
ProjectName: jsii.String("MyTestProject"),
BuildSpec: codebuild.BuildSpec_FromObject(map[string]interface{}{
"version": jsii.String("0.2"),
"phases": map[string]map[string][]*string{
"build": map[string][]*string{
"commands": []*string{
jsii.String("echo \"Hello, CodeBuild!\""),
},
},
},
}),
})
task := tasks.NewCodeBuildStartBuild(this, jsii.String("Task"), &CodeBuildStartBuildProps{
Project: codebuildProject,
IntegrationPattern: sfn.IntegrationPattern_RUN_JOB,
EnvironmentVariablesOverride: map[string]BuildEnvironmentVariable{
"ZONE": &BuildEnvironmentVariable{
"type": codebuild.BuildEnvironmentVariableType_PLAINTEXT,
"value": sfn.JsonPath_stringAt(jsii.String("$.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.
ts
import * as codebuild from 'aws-cdk-lib/aws-codebuild';
const project = new codebuild.Project(this, 'Project', {
projectName: 'MyTestProject',
buildSpec: codebuild.BuildSpec.fromObjectToYaml({
version: 0.2,
batch: {
'build-list': [
{
identifier: 'id',
buildspec: 'version: 0.2\nphases:\n build:\n commands:\n - echo "Hello, from small!"',
},
],
},
}),
});
project.enableBatchBuilds();
const task = new tasks.CodeBuildStartBuildBatch(this, 'buildBatchTask', {
project,
integrationPattern: sfn.IntegrationPattern.REQUEST_RESPONSE,
environmentVariablesOverride: {
test: {
type: codebuild.BuildEnvironmentVariableType.PLAINTEXT,
value: 'testValue',
},
},
});
python
import aws_cdk.aws_codebuild as codebuild
project = codebuild.Project(self, "Project",
project_name="MyTestProject",
build_spec=codebuild.BuildSpec.from_object_to_yaml({
"version": 0.2,
"batch": {
"build-list": [{
"identifier": "id",
"buildspec": "version: 0.2\nphases:\n build:\n commands:\n - echo \"Hello, from small!\""
}
]
}
})
)
project.enable_batch_builds()
task = tasks.CodeBuildStartBuildBatch(self, "buildBatchTask",
project=project,
integration_pattern=sfn.IntegrationPattern.REQUEST_RESPONSE,
environment_variables_override={
"test": codebuild.BuildEnvironmentVariable(
type=codebuild.BuildEnvironmentVariableType.PLAINTEXT,
value="testValue"
)
}
)
java
import software.amazon.awscdk.services.codebuild.*;
Project project = Project.Builder.create(this, "Project")
.projectName("MyTestProject")
.buildSpec(BuildSpec.fromObjectToYaml(Map.of(
"version", 0.2,
"batch", Map.of(
"build-list", List.of(Map.of(
"identifier", "id",
"buildspec", "version: 0.2\nphases:\n build:\n commands:\n - echo \"Hello, from small!\""))))))
.build();
project.enableBatchBuilds();
CodeBuildStartBuildBatch task = CodeBuildStartBuildBatch.Builder.create(this, "buildBatchTask")
.project(project)
.integrationPattern(IntegrationPattern.REQUEST_RESPONSE)
.environmentVariablesOverride(Map.of(
"test", BuildEnvironmentVariable.builder()
.type(BuildEnvironmentVariableType.PLAINTEXT)
.value("testValue")
.build()))
.build();
csharp
using Amazon.CDK.AWS.CodeBuild;
var project = new Project(this, "Project", new ProjectProps {
ProjectName = "MyTestProject",
BuildSpec = BuildSpec.FromObjectToYaml(new Dictionary<string, object> {
{ "version", 0.2 },
{ "batch", new Dictionary<string, IDictionary<string, string>[]> {
{ "build-list", new [] { new Struct {
Identifier = "id",
Buildspec = @"version: 0.2
phases:
build:
commands:
- echo ""Hello, from small!"""
} } }
} }
})
});
project.EnableBatchBuilds();
var task = new CodeBuildStartBuildBatch(this, "buildBatchTask", new CodeBuildStartBuildBatchProps {
Project = project,
IntegrationPattern = IntegrationPattern.REQUEST_RESPONSE,
EnvironmentVariablesOverride = new Dictionary<string, BuildEnvironmentVariable> {
{ "test", new BuildEnvironmentVariable {
Type = BuildEnvironmentVariableType.PLAINTEXT,
Value = "testValue"
} }
}
});
go
import "github.com/aws/aws-cdk-go/awscdk"
project := codebuild.NewProject(this, jsii.String("Project"), &ProjectProps{
ProjectName: jsii.String("MyTestProject"),
BuildSpec: codebuild.BuildSpec_FromObjectToYaml(map[string]interface{}{
"version": jsii.Number(0.2),
"batch": map[string][]map[string]*string{
"build-list": []map[string]*string{
map[string]*string{
"identifier": jsii.String("id"),
"buildspec": jsii.String("version: 0.2\nphases:\n build:\n commands:\n - echo \"Hello, from small!\""),
},
},
},
}),
})
project.EnableBatchBuilds()
task := tasks.NewCodeBuildStartBuildBatch(this, jsii.String("buildBatchTask"), &CodeBuildStartBuildBatchProps{
Project: Project,
IntegrationPattern: sfn.IntegrationPattern_REQUEST_RESPONSE,
EnvironmentVariablesOverride: map[string]BuildEnvironmentVariable{
"test": &BuildEnvironmentVariable{
"type": codebuild.BuildEnvironmentVariableType_PLAINTEXT,
"value": jsii.String("testValue"),
},
},
})
Note: enableBatchBuilds() 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.
ts
declare const myTable: dynamodb.Table;
new tasks.DynamoGetItem(this, 'Get Item', {
key: { messageId: tasks.DynamoAttributeValue.fromString('message-007') },
table: myTable,
});
python
# my_table: dynamodb.Table
tasks.DynamoGetItem(self, "Get Item",
key={"message_id": tasks.DynamoAttributeValue.from_string("message-007")},
table=my_table
)
java
Table myTable;
DynamoGetItem.Builder.create(this, "Get Item")
.key(Map.of("messageId", DynamoAttributeValue.fromString("message-007")))
.table(myTable)
.build();
csharp
Table myTable;
new DynamoGetItem(this, "Get Item", new DynamoGetItemProps {
Key = new Dictionary<string, DynamoAttributeValue> { { "messageId", DynamoAttributeValue.FromString("message-007") } },
Table = myTable
});
go
var myTable Table
tasks.NewDynamoGetItem(this, jsii.String("Get Item"), &DynamoGetItemProps{
Key: map[string]DynamoAttributeValue{
"messageId": tasks.DynamoAttributeValue_fromString(jsii.String("message-007")),
},
Table: myTable,
})
ts
declare const myTable: dynamodb.Table;
new tasks.DynamoPutItem(this, 'PutItem', {
item: {
MessageId: tasks.DynamoAttributeValue.fromString('message-007'),
Text: tasks.DynamoAttributeValue.fromString(sfn.JsonPath.stringAt('$.bar')),
TotalCount: tasks.DynamoAttributeValue.fromNumber(10),
},
table: myTable,
});
python
# my_table: dynamodb.Table
tasks.DynamoPutItem(self, "PutItem",
item={
"MessageId": tasks.DynamoAttributeValue.from_string("message-007"),
"Text": tasks.DynamoAttributeValue.from_string(sfn.JsonPath.string_at("$.bar")),
"TotalCount": tasks.DynamoAttributeValue.from_number(10)
},
table=my_table
)
java
Table myTable;
DynamoPutItem.Builder.create(this, "PutItem")
.item(Map.of(
"MessageId", DynamoAttributeValue.fromString("message-007"),
"Text", DynamoAttributeValue.fromString(JsonPath.stringAt("$.bar")),
"TotalCount", DynamoAttributeValue.fromNumber(10)))
.table(myTable)
.build();
csharp
Table myTable;
new DynamoPutItem(this, "PutItem", new DynamoPutItemProps {
Item = new Dictionary<string, DynamoAttributeValue> {
{ "MessageId", DynamoAttributeValue.FromString("message-007") },
{ "Text", DynamoAttributeValue.FromString(JsonPath.StringAt("$.bar")) },
{ "TotalCount", DynamoAttributeValue.FromNumber(10) }
},
Table = myTable
});
go
var myTable Table
tasks.NewDynamoPutItem(this, jsii.String("PutItem"), &DynamoPutItemProps{
Item: map[string]DynamoAttributeValue{
"MessageId": tasks.DynamoAttributeValue_fromString(jsii.String("message-007")),
"Text": tasks.DynamoAttributeValue_fromString(sfn.JsonPath_stringAt(jsii.String("$.bar"))),
"TotalCount": tasks.DynamoAttributeValue_fromNumber(jsii.Number(10)),
},
Table: myTable,
})
DeleteItem
The DeleteItem operation deletes a single item in a table by primary key.
ts
declare const myTable: dynamodb.Table;
new tasks.DynamoDeleteItem(this, 'DeleteItem', {
key: { MessageId: tasks.DynamoAttributeValue.fromString('message-007') },
table: myTable,
resultPath: sfn.JsonPath.DISCARD,
});
python
# my_table: dynamodb.Table
tasks.DynamoDeleteItem(self, "DeleteItem",
key={"MessageId": tasks.DynamoAttributeValue.from_string("message-007")},
table=my_table,
result_path=sfn.JsonPath.DISCARD
)
java
Table myTable;
DynamoDeleteItem.Builder.create(this, "DeleteItem")
.key(Map.of("MessageId", DynamoAttributeValue.fromString("message-007")))
.table(myTable)
.resultPath(JsonPath.DISCARD)
.build();
csharp
Table myTable;
new DynamoDeleteItem(this, "DeleteItem", new DynamoDeleteItemProps {
Key = new Dictionary<string, DynamoAttributeValue> { { "MessageId", DynamoAttributeValue.FromString("message-007") } },
Table = myTable,
ResultPath = JsonPath.DISCARD
});
go
var myTable Table
tasks.NewDynamoDeleteItem(this, jsii.String("DeleteItem"), &DynamoDeleteItemProps{
Key: map[string]DynamoAttributeValue{
"MessageId": tasks.DynamoAttributeValue_fromString(jsii.String("message-007")),
},
Table: myTable,
ResultPath: sfn.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.
ts
declare const myTable: dynamodb.Table;
new tasks.DynamoUpdateItem(this, 'UpdateItem', {
key: {
MessageId: tasks.DynamoAttributeValue.fromString('message-007')
},
table: myTable,
expressionAttributeValues: {
':val': tasks.DynamoAttributeValue.numberFromString(sfn.JsonPath.stringAt('$.Item.TotalCount.N')),
':rand': tasks.DynamoAttributeValue.fromNumber(20),
},
updateExpression: 'SET TotalCount = :val + :rand',
});
python
# my_table: dynamodb.Table
tasks.DynamoUpdateItem(self, "UpdateItem",
key={
"MessageId": tasks.DynamoAttributeValue.from_string("message-007")
},
table=my_table,
expression_attribute_values={
":val": tasks.DynamoAttributeValue.number_from_string(sfn.JsonPath.string_at("$.Item.TotalCount.N")),
":rand": tasks.DynamoAttributeValue.from_number(20)
},
update_expression="SET TotalCount = :val + :rand"
)
java
Table myTable;
DynamoUpdateItem.Builder.create(this, "UpdateItem")
.key(Map.of(
"MessageId", DynamoAttributeValue.fromString("message-007")))
.table(myTable)
.expressionAttributeValues(Map.of(
":val", DynamoAttributeValue.numberFromString(JsonPath.stringAt("$.Item.TotalCount.N")),
":rand", DynamoAttributeValue.fromNumber(20)))
.updateExpression("SET TotalCount = :val + :rand")
.build();
csharp
Table myTable;
new DynamoUpdateItem(this, "UpdateItem", new DynamoUpdateItemProps {
Key = new Dictionary<string, DynamoAttributeValue> {
{ "MessageId", DynamoAttributeValue.FromString("message-007") }
},
Table = myTable,
ExpressionAttributeValues = new Dictionary<string, DynamoAttributeValue> {
{ ":val", DynamoAttributeValue.NumberFromString(JsonPath.StringAt("$.Item.TotalCount.N")) },
{ ":rand", DynamoAttributeValue.FromNumber(20) }
},
UpdateExpression = "SET TotalCount = :val + :rand"
});
go
var myTable Table
tasks.NewDynamoUpdateItem(this, jsii.String("UpdateItem"), &DynamoUpdateItemProps{
Key: map[string]DynamoAttributeValue{
"MessageId": tasks.DynamoAttributeValue_fromString(jsii.String("message-007")),
},
Table: myTable,
ExpressionAttributeValues: map[string]DynamoAttributeValue{
":val": tasks.DynamoAttributeValue_numberFromString(sfn.JsonPath_stringAt(jsii.String("$.Item.TotalCount.N"))),
":rand": tasks.DynamoAttributeValue_fromNumber(jsii.Number(20)),
},
UpdateExpression: jsii.String("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
ts
const vpc = ec2.Vpc.fromLookup(this, 'Vpc', {
isDefault: true,
});
const cluster = new ecs.Cluster(this, 'Ec2Cluster', { vpc });
cluster.addCapacity('DefaultAutoScalingGroup', {
instanceType: new ec2.InstanceType('t2.micro'),
vpcSubnets: { subnetType: ec2.SubnetType.PUBLIC },
});
const taskDefinition = new ecs.TaskDefinition(this, 'TD', {
compatibility: ecs.Compatibility.EC2,
});
taskDefinition.addContainer('TheContainer', {
image: ecs.ContainerImage.fromRegistry('foo/bar'),
memoryLimitMiB: 256,
});
const runTask = new tasks.EcsRunTask(this, 'Run', {
integrationPattern: sfn.IntegrationPattern.RUN_JOB,
cluster,
taskDefinition,
launchTarget: new tasks.EcsEc2LaunchTarget({
placementStrategies: [
ecs.PlacementStrategy.spreadAcrossInstances(),
ecs.PlacementStrategy.packedByCpu(),
ecs.PlacementStrategy.randomly(),
],
placementConstraints: [
ecs.PlacementConstraint.memberOf('blieptuut'),
],
}),
propagatedTagSource: ecs.PropagatedTagSource.TASK_DEFINITION,
});
python
vpc = ec2.Vpc.from_lookup(self, "Vpc",
is_default=True
)
cluster = ecs.Cluster(self, "Ec2Cluster", vpc=vpc)
cluster.add_capacity("DefaultAutoScalingGroup",
instance_type=ec2.InstanceType("t2.micro"),
vpc_subnets=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PUBLIC)
)
task_definition = ecs.TaskDefinition(self, "TD",
compatibility=ecs.Compatibility.EC2
)
task_definition.add_container("TheContainer",
image=ecs.ContainerImage.from_registry("foo/bar"),
memory_limit_mi_b=256
)
run_task = tasks.EcsRunTask(self, "Run",
integration_pattern=sfn.IntegrationPattern.RUN_JOB,
cluster=cluster,
task_definition=task_definition,
launch_target=tasks.EcsEc2LaunchTarget(
placement_strategies=[
ecs.PlacementStrategy.spread_across_instances(),
ecs.PlacementStrategy.packed_by_cpu(),
ecs.PlacementStrategy.randomly()
],
placement_constraints=[
ecs.PlacementConstraint.member_of("blieptuut")
]
),
propagated_tag_source=ecs.PropagatedTagSource.TASK_DEFINITION
)
java
IVpc vpc = Vpc.fromLookup(this, "Vpc", VpcLookupOptions.builder()
.isDefault(true)
.build());
Cluster cluster = Cluster.Builder.create(this, "Ec2Cluster").vpc(vpc).build();
cluster.addCapacity("DefaultAutoScalingGroup", AddCapacityOptions.builder()
.instanceType(new InstanceType("t2.micro"))
.vpcSubnets(SubnetSelection.builder().subnetType(SubnetType.PUBLIC).build())
.build());
TaskDefinition taskDefinition = TaskDefinition.Builder.create(this, "TD")
.compatibility(Compatibility.EC2)
.build();
taskDefinition.addContainer("TheContainer", ContainerDefinitionOptions.builder()
.image(ContainerImage.fromRegistry("foo/bar"))
.memoryLimitMiB(256)
.build());
EcsRunTask runTask = EcsRunTask.Builder.create(this, "Run")
.integrationPattern(IntegrationPattern.RUN_JOB)
.cluster(cluster)
.taskDefinition(taskDefinition)
.launchTarget(EcsEc2LaunchTarget.Builder.create()
.placementStrategies(List.of(PlacementStrategy.spreadAcrossInstances(), PlacementStrategy.packedByCpu(), PlacementStrategy.randomly()))
.placementConstraints(List.of(PlacementConstraint.memberOf("blieptuut")))
.build())
.propagatedTagSource(PropagatedTagSource.TASK_DEFINITION)
.build();
csharp
var vpc = Vpc.FromLookup(this, "Vpc", new VpcLookupOptions {
IsDefault = true
});
var cluster = new Cluster(this, "Ec2Cluster", new ClusterProps { Vpc = vpc });
cluster.AddCapacity("DefaultAutoScalingGroup", new AddCapacityOptions {
InstanceType = new InstanceType("t2.micro"),
VpcSubnets = new SubnetSelection { SubnetType = SubnetType.PUBLIC }
});
var taskDefinition = new TaskDefinition(this, "TD", new TaskDefinitionProps {
Compatibility = Compatibility.EC2
});
taskDefinition.AddContainer("TheContainer", new ContainerDefinitionOptions {
Image = ContainerImage.FromRegistry("foo/bar"),
MemoryLimitMiB = 256
});
var runTask = new EcsRunTask(this, "Run", new EcsRunTaskProps {
IntegrationPattern = IntegrationPattern.RUN_JOB,
Cluster = cluster,
TaskDefinition = taskDefinition,
LaunchTarget = new EcsEc2LaunchTarget(new EcsEc2LaunchTargetOptions {
PlacementStrategies = new [] { PlacementStrategy.SpreadAcrossInstances(), PlacementStrategy.PackedByCpu(), PlacementStrategy.Randomly() },
PlacementConstraints = new [] { PlacementConstraint.MemberOf("blieptuut") }
}),
PropagatedTagSource = PropagatedTagSource.TASK_DEFINITION
});
go
vpc := ec2.Vpc_FromLookup(this, jsii.String("Vpc"), &VpcLookupOptions{
IsDefault: jsii.Boolean(true),
})
cluster := ecs.NewCluster(this, jsii.String("Ec2Cluster"), &ClusterProps{
Vpc: Vpc,
})
cluster.AddCapacity(jsii.String("DefaultAutoScalingGroup"), &AddCapacityOptions{
InstanceType: ec2.NewInstanceType(jsii.String("t2.micro")),
VpcSubnets: &SubnetSelection{
SubnetType: ec2.SubnetType_PUBLIC,
},
})
taskDefinition := ecs.NewTaskDefinition(this, jsii.String("TD"), &TaskDefinitionProps{
Compatibility: ecs.Compatibility_EC2,
})
taskDefinition.AddContainer(jsii.String("TheContainer"), &ContainerDefinitionOptions{
Image: ecs.ContainerImage_FromRegistry(jsii.String("foo/bar")),
MemoryLimitMiB: jsii.Number(256),
})
runTask := tasks.NewEcsRunTask(this, jsii.String("Run"), &EcsRunTaskProps{
IntegrationPattern: sfn.IntegrationPattern_RUN_JOB,
Cluster: Cluster,
TaskDefinition: TaskDefinition,
LaunchTarget: tasks.NewEcsEc2LaunchTarget(&EcsEc2LaunchTargetOptions{
PlacementStrategies: []PlacementStrategy{
ecs.PlacementStrategy_SpreadAcrossInstances(),
ecs.PlacementStrategy_PackedByCpu(),
ecs.PlacementStrategy_Randomly(),
},
PlacementConstraints: []PlacementConstraint{
ecs.PlacementConstraint_MemberOf(jsii.String("blieptuut")),
},
}),
PropagatedTagSource: 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
ts
const vpc = ec2.Vpc.fromLookup(this, 'Vpc', {
isDefault: true,
});
const cluster = new ecs.Cluster(this, 'FargateCluster', { vpc });
const taskDefinition = new ecs.TaskDefinition(this, 'TD', {
memoryMiB: '512',
cpu: '256',
compatibility: ecs.Compatibility.FARGATE,
});
const containerDefinition = taskDefinition.addContainer('TheContainer', {
image: ecs.ContainerImage.fromRegistry('foo/bar'),
memoryLimitMiB: 256,
});
const runTask = new tasks.EcsRunTask(this, 'RunFargate', {
integrationPattern: sfn.IntegrationPattern.RUN_JOB,
cluster,
taskDefinition,
assignPublicIp: true,
containerOverrides: [{
containerDefinition,
environment: [{ name: 'SOME_KEY', value: sfn.JsonPath.stringAt('$.SomeKey') }],
}],
launchTarget: new tasks.EcsFargateLaunchTarget(),
propagatedTagSource: ecs.PropagatedTagSource.TASK_DEFINITION,
});
python
vpc = ec2.Vpc.from_lookup(self, "Vpc",
is_default=True
)
cluster = ecs.Cluster(self, "FargateCluster", vpc=vpc)
task_definition = ecs.TaskDefinition(self, "TD",
memory_mi_b="512",
cpu="256",
compatibility=ecs.Compatibility.FARGATE
)
container_definition = task_definition.add_container("TheContainer",
image=ecs.ContainerImage.from_registry("foo/bar"),
memory_limit_mi_b=256
)
run_task = tasks.EcsRunTask(self, "RunFargate",
integration_pattern=sfn.IntegrationPattern.RUN_JOB,
cluster=cluster,
task_definition=task_definition,
assign_public_ip=True,
container_overrides=[tasks.ContainerOverride(
container_definition=container_definition,
environment=[tasks.TaskEnvironmentVariable(name="SOME_KEY", value=sfn.JsonPath.string_at("$.SomeKey"))]
)],
launch_target=tasks.EcsFargateLaunchTarget(),
propagated_tag_source=ecs.PropagatedTagSource.TASK_DEFINITION
)
java
IVpc vpc = Vpc.fromLookup(this, "Vpc", VpcLookupOptions.builder()
.isDefault(true)
.build());
Cluster cluster = Cluster.Builder.create(this, "FargateCluster").vpc(vpc).build();
TaskDefinition taskDefinition = TaskDefinition.Builder.create(this, "TD")
.memoryMiB("512")
.cpu("256")
.compatibility(Compatibility.FARGATE)
.build();
ContainerDefinition containerDefinition = taskDefinition.addContainer("TheContainer", ContainerDefinitionOptions.builder()
.image(ContainerImage.fromRegistry("foo/bar"))
.memoryLimitMiB(256)
.build());
EcsRunTask runTask = EcsRunTask.Builder.create(this, "RunFargate")
.integrationPattern(IntegrationPattern.RUN_JOB)
.cluster(cluster)
.taskDefinition(taskDefinition)
.assignPublicIp(true)
.containerOverrides(List.of(ContainerOverride.builder()
.containerDefinition(containerDefinition)
.environment(List.of(TaskEnvironmentVariable.builder().name("SOME_KEY").value(JsonPath.stringAt("$.SomeKey")).build()))
.build()))
.launchTarget(new EcsFargateLaunchTarget())
.propagatedTagSource(PropagatedTagSource.TASK_DEFINITION)
.build();
csharp
var vpc = Vpc.FromLookup(this, "Vpc", new VpcLookupOptions {
IsDefault = true
});
var cluster = new Cluster(this, "FargateCluster", new ClusterProps { Vpc = vpc });
var taskDefinition = new TaskDefinition(this, "TD", new TaskDefinitionProps {
MemoryMiB = "512",
Cpu = "256",
Compatibility = Compatibility.FARGATE
});
var containerDefinition = taskDefinition.AddContainer("TheContainer", new ContainerDefinitionOptions {
Image = ContainerImage.FromRegistry("foo/bar"),
MemoryLimitMiB = 256
});
var runTask = new EcsRunTask(this, "RunFargate", new EcsRunTaskProps {
IntegrationPattern = IntegrationPattern.RUN_JOB,
Cluster = cluster,
TaskDefinition = taskDefinition,
AssignPublicIp = true,
ContainerOverrides = new [] { new ContainerOverride {
ContainerDefinition = containerDefinition,
Environment = new [] { new TaskEnvironmentVariable { Name = "SOME_KEY", Value = JsonPath.StringAt("$.SomeKey") } }
} },
LaunchTarget = new EcsFargateLaunchTarget(),
PropagatedTagSource = PropagatedTagSource.TASK_DEFINITION
});
go
vpc := ec2.Vpc_FromLookup(this, jsii.String("Vpc"), &VpcLookupOptions{
IsDefault: jsii.Boolean(true),
})
cluster := ecs.NewCluster(this, jsii.String("FargateCluster"), &ClusterProps{
Vpc: Vpc,
})
taskDefinition := ecs.NewTaskDefinition(this, jsii.String("TD"), &TaskDefinitionProps{
MemoryMiB: jsii.String("512"),
Cpu: jsii.String("256"),
Compatibility: ecs.Compatibility_FARGATE,
})
containerDefinition := taskDefinition.AddContainer(jsii.String("TheContainer"), &ContainerDefinitionOptions{
Image: ecs.ContainerImage_FromRegistry(jsii.String("foo/bar")),
MemoryLimitMiB: jsii.Number(256),
})
runTask := tasks.NewEcsRunTask(this, jsii.String("RunFargate"), &EcsRunTaskProps{
IntegrationPattern: sfn.IntegrationPattern_RUN_JOB,
Cluster: Cluster,
TaskDefinition: TaskDefinition,
AssignPublicIp: jsii.Boolean(true),
ContainerOverrides: []ContainerOverride{
&ContainerOverride{
ContainerDefinition: *ContainerDefinition,
Environment: []TaskEnvironmentVariable{
&TaskEnvironmentVariable{
Name: jsii.String("SOME_KEY"),
Value: sfn.JsonPath_StringAt(jsii.String("$.SomeKey")),
},
},
},
},
LaunchTarget: tasks.NewEcsFargateLaunchTarget(),
PropagatedTagSource: ecs.PropagatedTagSource_TASK_DEFINITION,
})
Capacity Provider Options
The capacityProviderOptions 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
capacityProviderOptionsis not specified, the task uses the launch type (EC2 or FARGATE) without a capacity provider strategy.
ts
const vpc = ec2.Vpc.fromLookup(this, 'Vpc', {
isDefault: true,
});
const cluster = new ecs.Cluster(this, 'FargateCluster', { vpc });
const taskDefinition = new ecs.TaskDefinition(this, 'TD', {
memoryMiB: '512',
cpu: '256',
compatibility: ecs.Compatibility.FARGATE,
});
// Use custom() option - specify custom capacity provider strategy
const runTaskWithCustom = new tasks.EcsRunTask(this, 'RunTaskWithCustom', {
cluster,
taskDefinition,
launchTarget: new tasks.EcsFargateLaunchTarget({
platformVersion: ecs.FargatePlatformVersion.VERSION1_4,
capacityProviderOptions: tasks.CapacityProviderOptions.custom([
{ capacityProvider: 'FARGATE_SPOT', weight: 2, base: 1 },
{ capacityProvider: 'FARGATE', weight: 1 },
]),
}),
});
// Use default() option - uses cluster's default capacity provider strategy
const runTaskWithDefault = new tasks.EcsRunTask(this, 'RunTaskWithDefault', {
cluster,
taskDefinition,
launchTarget: new tasks.EcsFargateLaunchTarget({
platformVersion: ecs.FargatePlatformVersion.VERSION1_4,
capacityProviderOptions: tasks.CapacityProviderOptions.default(),
}),
});
python
vpc = ec2.Vpc.from_lookup(self, "Vpc",
is_default=True
)
cluster = ecs.Cluster(self, "FargateCluster", vpc=vpc)
task_definition = ecs.TaskDefinition(self, "TD",
memory_mi_b="512",
cpu="256",
compatibility=ecs.Compatibility.FARGATE
)
# Use custom() option - specify custom capacity provider strategy
run_task_with_custom = tasks.EcsRunTask(self, "RunTaskWithCustom",
cluster=cluster,
task_definition=task_definition,
launch_target=tasks.EcsFargateLaunchTarget(
platform_version=ecs.FargatePlatformVersion.VERSION1_4,
capacity_provider_options=tasks.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 = tasks.EcsRunTask(self, "RunTaskWithDefault",
cluster=cluster,
task_definition=task_definition,
launch_target=tasks.EcsFargateLaunchTarget(
platform_version=ecs.FargatePlatformVersion.VERSION1_4,
capacity_provider_options=tasks.CapacityProviderOptions.default()
)
)
java
IVpc vpc = Vpc.fromLookup(this, "Vpc", VpcLookupOptions.builder()
.isDefault(true)
.build());
Cluster cluster = Cluster.Builder.create(this, "FargateCluster").vpc(vpc).build();
TaskDefinition taskDefinition = TaskDefinition.Builder.create(this, "TD")
.memoryMiB("512")
.cpu("256")
.compatibility(Compatibility.FARGATE)
.build();
// Use custom() option - specify custom capacity provider strategy
EcsRunTask runTaskWithCustom = EcsRunTask.Builder.create(this, "RunTaskWithCustom")
.cluster(cluster)
.taskDefinition(taskDefinition)
.launchTarget(EcsFargateLaunchTarget.Builder.create()
.platformVersion(FargatePlatformVersion.VERSION1_4)
.capacityProviderOptions(CapacityProviderOptions.custom(List.of(CapacityProviderStrategy.builder().capacityProvider("FARGATE_SPOT").weight(2).base(1).build(), CapacityProviderStrategy.builder().capacityProvider("FARGATE").weight(1).build())))
.build())
.build();
// Use default() option - uses cluster's default capacity provider strategy
EcsRunTask runTaskWithDefault = EcsRunTask.Builder.create(this, "RunTaskWithDefault")
.cluster(cluster)
.taskDefinition(taskDefinition)
.launchTarget(EcsFargateLaunchTarget.Builder.create()
.platformVersion(FargatePlatformVersion.VERSION1_4)
.capacityProviderOptions(CapacityProviderOptions.default())
.build())
.build();
csharp
var vpc = Vpc.FromLookup(this, "Vpc", new VpcLookupOptions {
IsDefault = true
});
var cluster = new Cluster(this, "FargateCluster", new ClusterProps { Vpc = vpc });
var taskDefinition = new TaskDefinition(this, "TD", new TaskDefinitionProps {
MemoryMiB = "512",
Cpu = "256",
Compatibility = Compatibility.FARGATE
});
// Use custom() option - specify custom capacity provider strategy
var runTaskWithCustom = new EcsRunTask(this, "RunTaskWithCustom", new EcsRunTaskProps {
Cluster = cluster,
TaskDefinition = taskDefinition,
LaunchTarget = new EcsFargateLaunchTarget(new EcsFargateLaunchTargetOptions {
PlatformVersion = FargatePlatformVersion.VERSION1_4,
CapacityProviderOptions = CapacityProviderOptions.Custom(new [] { new CapacityProviderStrategy { CapacityProvider = "FARGATE_SPOT", Weight = 2, Base = 1 }, new CapacityProviderStrategy { CapacityProvider = "FARGATE", Weight = 1 } })
})
});
// Use default() option - uses cluster's default capacity provider strategy
var runTaskWithDefault = new EcsRunTask(this, "RunTaskWithDefault", new EcsRunTaskProps {
Cluster = cluster,
TaskDefinition = taskDefinition,
LaunchTarget = new EcsFargateLaunchTarget(new EcsFargateLaunchTargetOptions {
PlatformVersion = FargatePlatformVersion.VERSION1_4,
CapacityProviderOptions = CapacityProviderOptions.Default()
})
});
go
vpc := ec2.Vpc_FromLookup(this, jsii.String("Vpc"), &VpcLookupOptions{
IsDefault: jsii.Boolean(true),
})
cluster := ecs.NewCluster(this, jsii.String("FargateCluster"), &ClusterProps{
Vpc: Vpc,
})
taskDefinition := ecs.NewTaskDefinition(this, jsii.String("TD"), &TaskDefinitionProps{
MemoryMiB: jsii.String("512"),
Cpu: jsii.String("256"),
Compatibility: ecs.Compatibility_FARGATE,
})
// Use custom() option - specify custom capacity provider strategy
runTaskWithCustom := tasks.NewEcsRunTask(this, jsii.String("RunTaskWithCustom"), &EcsRunTaskProps{
Cluster: Cluster,
TaskDefinition: TaskDefinition,
LaunchTarget: tasks.NewEcsFargateLaunchTarget(&EcsFargateLaunchTargetOptions{
PlatformVersion: ecs.FargatePlatformVersion_VERSION1_4,
CapacityProviderOptions: tasks.CapacityProviderOptions_Custom([]CapacityProviderStrategy{
&CapacityProviderStrategy{
CapacityProvider: jsii.String("FARGATE_SPOT"),
Weight: jsii.Number(2),
Base: jsii.Number(1),
},
&CapacityProviderStrategy{
CapacityProvider: jsii.String("FARGATE"),
Weight: jsii.Number(1),
},
}),
}),
})
// Use default() option - uses cluster's default capacity provider strategy
runTaskWithDefault := tasks.NewEcsRunTask(this, jsii.String("RunTaskWithDefault"), &EcsRunTaskProps{
Cluster: Cluster,
TaskDefinition: TaskDefinition,
LaunchTarget: tasks.NewEcsFargateLaunchTarget(&EcsFargateLaunchTargetOptions{
PlatformVersion: ecs.FargatePlatformVersion_VERSION1_4,
CapacityProviderOptions: tasks.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
ts
const vpc = ec2.Vpc.fromLookup(this, 'Vpc', {
isDefault: true,
});
const cluster = new ecs.Cluster(this, 'ECSCluster', { vpc });
const taskDefinition = new ecs.TaskDefinition(this, 'TD', {
compatibility: ecs.Compatibility.FARGATE,
cpu: '256',
memoryMiB: '512'
});
taskDefinition.addContainer('TheContainer', {
image: ecs.ContainerImage.fromRegistry('foo/bar'),
});
const runTask = new tasks.EcsRunTask(this, 'Run', {
integrationPattern: sfn.IntegrationPattern.RUN_JOB,
cluster,
taskDefinition,
launchTarget: new tasks.EcsFargateLaunchTarget(),
cpu: '1024',
memoryMiB: '1048'
});
python
vpc = ec2.Vpc.from_lookup(self, "Vpc",
is_default=True
)
cluster = ecs.Cluster(self, "ECSCluster", vpc=vpc)
task_definition = ecs.TaskDefinition(self, "TD",
compatibility=ecs.Compatibility.FARGATE,
cpu="256",
memory_mi_b="512"
)
task_definition.add_container("TheContainer",
image=ecs.ContainerImage.from_registry("foo/bar")
)
run_task = tasks.EcsRunTask(self, "Run",
integration_pattern=sfn.IntegrationPattern.RUN_JOB,
cluster=cluster,
task_definition=task_definition,
launch_target=tasks.EcsFargateLaunchTarget(),
cpu="1024",
memory_mi_b="1048"
)
java
IVpc vpc = Vpc.fromLookup(this, "Vpc", VpcLookupOptions.builder()
.isDefault(true)
.build());
Cluster cluster = Cluster.Builder.create(this, "ECSCluster").vpc(vpc).build();
TaskDefinition taskDefinition = TaskDefinition.Builder.create(this, "TD")
.compatibility(Compatibility.FARGATE)
.cpu("256")
.memoryMiB("512")
.build();
taskDefinition.addContainer("TheContainer", ContainerDefinitionOptions.builder()
.image(ContainerImage.fromRegistry("foo/bar"))
.build());
EcsRunTask runTask = EcsRunTask.Builder.create(this, "Run")
.integrationPattern(IntegrationPattern.RUN_JOB)
.cluster(cluster)
.taskDefinition(taskDefinition)
.launchTarget(new EcsFargateLaunchTarget())
.cpu("1024")
.memoryMiB("1048")
.build();
csharp
var vpc = Vpc.FromLookup(this, "Vpc", new VpcLookupOptions {
IsDefault = true
});
var cluster = new Cluster(this, "ECSCluster", new ClusterProps { Vpc = vpc });
var taskDefinition = new TaskDefinition(this, "TD", new TaskDefinitionProps {
Compatibility = Compatibility.FARGATE,
Cpu = "256",
MemoryMiB = "512"
});
taskDefinition.AddContainer("TheContainer", new ContainerDefinitionOptions {
Image = ContainerImage.FromRegistry("foo/bar")
});
var runTask = new EcsRunTask(this, "Run", new EcsRunTaskProps {
IntegrationPattern = IntegrationPattern.RUN_JOB,
Cluster = cluster,
TaskDefinition = taskDefinition,
LaunchTarget = new EcsFargateLaunchTarget(),
Cpu = "1024",
MemoryMiB = "1048"
});
go
vpc := ec2.Vpc_FromLookup(this, jsii.String("Vpc"), &VpcLookupOptions{
IsDefault: jsii.Boolean(true),
})
cluster := ecs.NewCluster(this, jsii.String("ECSCluster"), &ClusterProps{
Vpc: Vpc,
})
taskDefinition := ecs.NewTaskDefinition(this, jsii.String("TD"), &TaskDefinitionProps{
Compatibility: ecs.Compatibility_FARGATE,
Cpu: jsii.String("256"),
MemoryMiB: jsii.String("512"),
})
taskDefinition.AddContainer(jsii.String("TheContainer"), &ContainerDefinitionOptions{
Image: ecs.ContainerImage_FromRegistry(jsii.String("foo/bar")),
})
runTask := tasks.NewEcsRunTask(this, jsii.String("Run"), &EcsRunTaskProps{
IntegrationPattern: sfn.IntegrationPattern_RUN_JOB,
Cluster: Cluster,
TaskDefinition: TaskDefinition,
LaunchTarget: tasks.NewEcsFargateLaunchTarget(),
Cpu: jsii.String("1024"),
MemoryMiB: jsii.String("1048"),
})
ECS enable Exec
By setting the property enableExecuteCommand to true, you can enable the ECS Exec feature for the task for either Fargate or EC2 launch types.
ts
const vpc = ec2.Vpc.fromLookup(this, 'Vpc', {
isDefault: true,
});
const cluster = new ecs.Cluster(this, 'ECSCluster', { vpc });
const taskDefinition = new ecs.TaskDefinition(this, 'TD', {
compatibility: ecs.Compatibility.EC2,
});
taskDefinition.addContainer('TheContainer', {
image: ecs.ContainerImage.fromRegistry('foo/bar'),
memoryLimitMiB: 256,
});
const runTask = new tasks.EcsRunTask(this, 'Run', {
integrationPattern: sfn.IntegrationPattern.RUN_JOB,
cluster,
taskDefinition,
launchTarget: new tasks.EcsEc2LaunchTarget(),
enableExecuteCommand: true,
});
python
vpc = ec2.Vpc.from_lookup(self, "Vpc",
is_default=True
)
cluster = ecs.Cluster(self, "ECSCluster", vpc=vpc)
task_definition = ecs.TaskDefinition(self, "TD",
compatibility=ecs.Compatibility.EC2
)
task_definition.add_container("TheContainer",
image=ecs.ContainerImage.from_registry("foo/bar"),
memory_limit_mi_b=256
)
run_task = tasks.EcsRunTask(self, "Run",
integration_pattern=sfn.IntegrationPattern.RUN_JOB,
cluster=cluster,
task_definition=task_definition,
launch_target=tasks.EcsEc2LaunchTarget(),
enable_execute_command=True
)
java
IVpc vpc = Vpc.fromLookup(this, "Vpc", VpcLookupOptions.builder()
.isDefault(true)
.build());
Cluster cluster = Cluster.Builder.create(this, "ECSCluster").vpc(vpc).build();
TaskDefinition taskDefinition = TaskDefinition.Builder.create(this, "TD")
.compatibility(Compatibility.EC2)
.build();
taskDefinition.addContainer("TheContainer", ContainerDefinitionOptions.builder()
.image(ContainerImage.fromRegistry("foo/bar"))
.memoryLimitMiB(256)
.build());
EcsRunTask runTask = EcsRunTask.Builder.create(this, "Run")
.integrationPattern(IntegrationPattern.RUN_JOB)
.cluster(cluster)
.taskDefinition(taskDefinition)
.launchTarget(new EcsEc2LaunchTarget())
.enableExecuteCommand(true)
.build();
csharp
var vpc = Vpc.FromLookup(this, "Vpc", new VpcLookupOptions {
IsDefault = true
});
var cluster = new Cluster(this, "ECSCluster", new ClusterProps { Vpc = vpc });
var taskDefinition = new TaskDefinition(this, "TD", new TaskDefinitionProps {
Compatibility = Compatibility.EC2
});
taskDefinition.AddContainer("TheContainer", new ContainerDefinitionOptions {
Image = ContainerImage.FromRegistry("foo/bar"),
MemoryLimitMiB = 256
});
var runTask = new EcsRunTask(this, "Run", new EcsRunTaskProps {
IntegrationPattern = IntegrationPattern.RUN_JOB,
Cluster = cluster,
TaskDefinition = taskDefinition,
LaunchTarget = new EcsEc2LaunchTarget(),
EnableExecuteCommand = true
});
go
vpc := ec2.Vpc_FromLookup(this, jsii.String("Vpc"), &VpcLookupOptions{
IsDefault: jsii.Boolean(true),
})
cluster := ecs.NewCluster(this, jsii.String("ECSCluster"), &ClusterProps{
Vpc: Vpc,
})
taskDefinition := ecs.NewTaskDefinition(this, jsii.String("TD"), &TaskDefinitionProps{
Compatibility: ecs.Compatibility_EC2,
})
taskDefinition.AddContainer(jsii.String("TheContainer"), &ContainerDefinitionOptions{
Image: ecs.ContainerImage_FromRegistry(jsii.String("foo/bar")),
MemoryLimitMiB: jsii.Number(256),
})
runTask := tasks.NewEcsRunTask(this, jsii.String("Run"), &EcsRunTaskProps{
IntegrationPattern: sfn.IntegrationPattern_RUN_JOB,
Cluster: Cluster,
TaskDefinition: TaskDefinition,
LaunchTarget: tasks.NewEcsEc2LaunchTarget(),
EnableExecuteCommand: jsii.Boolean(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 runJobFlow API in EMR.
ts
const clusterRole = new iam.Role(this, 'ClusterRole', {
assumedBy: new iam.ServicePrincipal('ec2.amazonaws.com'),
});
const serviceRole = new iam.Role(this, 'ServiceRole', {
assumedBy: new iam.ServicePrincipal('elasticmapreduce.amazonaws.com'),
});
const autoScalingRole = new iam.Role(this, 'AutoScalingRole', {
assumedBy: new iam.ServicePrincipal('elasticmapreduce.amazonaws.com'),
});
autoScalingRole.assumeRolePolicy?.addStatements(
new iam.PolicyStatement({
effect: iam.Effect.ALLOW,
principals: [
new iam.ServicePrincipal('application-autoscaling.amazonaws.com'),
],
actions: [
'sts:AssumeRole',
],
}));
)
new tasks.EmrCreateCluster(this, 'Create Cluster', {
instances: {},
clusterRole,
name: sfn.TaskInput.fromJsonPathAt('$.ClusterName').value,
serviceRole,
autoScalingRole,
});
python
cluster_role = iam.Role(self, "ClusterRole",
assumed_by=iam.ServicePrincipal("ec2.amazonaws.com")
)
service_role = iam.Role(self, "ServiceRole",
assumed_by=iam.ServicePrincipal("elasticmapreduce.amazonaws.com")
)
auto_scaling_role = iam.Role(self, "AutoScalingRole",
assumed_by=iam.ServicePrincipal("elasticmapreduce.amazonaws.com")
)
auto_scaling_role.assume_role_policy.add_statements(
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
principals=[
iam.ServicePrincipal("application-autoscaling.amazonaws.com")
],
actions=["sts:AssumeRole"
]
))
tasks.EmrCreateCluster(self, "Create Cluster",
instances=tasks.EmrCreateCluster.InstancesConfigProperty(),
cluster_role=cluster_role,
name=sfn.TaskInput.from_json_path_at("$.ClusterName").value,
service_role=service_role,
auto_scaling_role=auto_scaling_role
)
java
Role clusterRole = Role.Builder.create(this, "ClusterRole")
.assumedBy(new ServicePrincipal("ec2.amazonaws.com"))
.build();
Role serviceRole = Role.Builder.create(this, "ServiceRole")
.assumedBy(new ServicePrincipal("elasticmapreduce.amazonaws.com"))
.build();
Role autoScalingRole = Role.Builder.create(this, "AutoScalingRole")
.assumedBy(new ServicePrincipal("elasticmapreduce.amazonaws.com"))
.build();
autoScalingRole.assumeRolePolicy.addStatements(
PolicyStatement.Builder.create()
.effect(Effect.ALLOW)
.principals(List.of(
new ServicePrincipal("application-autoscaling.amazonaws.com")))
.actions(List.of("sts:AssumeRole"))
.build());
EmrCreateCluster.Builder.create(this, "Create Cluster")
.instances(InstancesConfigProperty.builder().build())
.clusterRole(clusterRole)
.name(TaskInput.fromJsonPathAt("$.ClusterName").getValue())
.serviceRole(serviceRole)
.autoScalingRole(autoScalingRole)
.build();
csharp
var clusterRole = new Role(this, "ClusterRole", new RoleProps {
AssumedBy = new ServicePrincipal("ec2.amazonaws.com")
});
var serviceRole = new Role(this, "ServiceRole", new RoleProps {
AssumedBy = new ServicePrincipal("elasticmapreduce.amazonaws.com")
});
var autoScalingRole = new Role(this, "AutoScalingRole", new RoleProps {
AssumedBy = new ServicePrincipal("elasticmapreduce.amazonaws.com")
});
autoScalingRole.AssumeRolePolicy.AddStatements(
new PolicyStatement(new PolicyStatementProps {
Effect = Effect.ALLOW,
Principals = new [] {
new ServicePrincipal("application-autoscaling.amazonaws.com") },
Actions = new [] { "sts:AssumeRole" }
}));
new EmrCreateCluster(this, "Create Cluster", new EmrCreateClusterProps {
Instances = new InstancesConfigProperty { },
ClusterRole = clusterRole,
Name = TaskInput.FromJsonPathAt("$.ClusterName").Value,
ServiceRole = serviceRole,
AutoScalingRole = autoScalingRole
});
go
clusterRole := iam.NewRole(this, jsii.String("ClusterRole"), &RoleProps{
AssumedBy: iam.NewServicePrincipal(jsii.String("ec2.amazonaws.com")),
})
serviceRole := iam.NewRole(this, jsii.String("ServiceRole"), &RoleProps{
AssumedBy: iam.NewServicePrincipal(jsii.String("elasticmapreduce.amazonaws.com")),
})
autoScalingRole := iam.NewRole(this, jsii.String("AutoScalingRole"), &RoleProps{
AssumedBy: iam.NewServicePrincipal(jsii.String("elasticmapreduce.amazonaws.com")),
})
autoScalingRole.AssumeRolePolicy.AddStatements(
iam.NewPolicyStatement(&PolicyStatementProps{
Effect: iam.Effect_ALLOW,
Principals: []IPrincipal{
iam.NewServicePrincipal(jsii.String("application-autoscaling.amazonaws.com")),
},
Actions: []*string{
jsii.String("sts:AssumeRole"),
},
}))
tasks.NewEmrCreateCluster(this, jsii.String("Create Cluster"), &EmrCreateClusterProps{
Instances: &InstancesConfigProperty{
},
ClusterRole: ClusterRole,
Name: sfn.TaskInput_FromJsonPathAt(jsii.String("$.ClusterName")).value,
ServiceRole: ServiceRole,
AutoScalingRole: AutoScalingRole,
})
You can use the launch specification for On-Demand and Spot instances in the fleet.
ts
new tasks.EmrCreateCluster(this, 'OnDemandSpecification', {
instances: {
instanceFleets: [{
instanceFleetType: tasks.EmrCreateCluster.InstanceRoleType.MASTER,
launchSpecifications: {
onDemandSpecification: {
allocationStrategy: tasks.EmrCreateCluster.OnDemandAllocationStrategy.LOWEST_PRICE,
},
},
}],
},
name: 'OnDemandCluster',
integrationPattern: sfn.IntegrationPattern.RUN_JOB,
});
new tasks.EmrCreateCluster(this, 'SpotSpecification', {
instances: {
instanceFleets: [{
instanceFleetType: tasks.EmrCreateCluster.InstanceRoleType.MASTER,
launchSpecifications: {
spotSpecification: {
allocationStrategy: tasks.EmrCreateCluster.SpotAllocationStrategy.CAPACITY_OPTIMIZED,
timeoutAction: tasks.EmrCreateCluster.SpotTimeoutAction.TERMINATE_CLUSTER,
timeout: Duration.minutes(5),
},
},
}],
},
name: 'SpotCluster',
integrationPattern: sfn.IntegrationPattern.RUN_JOB,
});
python
tasks.EmrCreateCluster(self, "OnDemandSpecification",
instances=tasks.EmrCreateCluster.InstancesConfigProperty(
instance_fleets=[tasks.EmrCreateCluster.InstanceFleetConfigProperty(
instance_fleet_type=tasks.EmrCreateCluster.InstanceRoleType.MASTER,
launch_specifications=tasks.EmrCreateCluster.InstanceFleetProvisioningSpecificationsProperty(
on_demand_specification=tasks.EmrCreateCluster.OnDemandProvisioningSpecificationProperty(
allocation_strategy=tasks.EmrCreateCluster.OnDemandAllocationStrategy.LOWEST_PRICE
)
)
)]
),
name="OnDemandCluster",
integration_pattern=sfn.IntegrationPattern.RUN_JOB
)
tasks.EmrCreateCluster(self, "SpotSpecification",
instances=tasks.EmrCreateCluster.InstancesConfigProperty(
instance_fleets=[tasks.EmrCreateCluster.InstanceFleetConfigProperty(
instance_fleet_type=tasks.EmrCreateCluster.InstanceRoleType.MASTER,
launch_specifications=tasks.EmrCreateCluster.InstanceFleetProvisioningSpecificationsProperty(
spot_specification=tasks.EmrCreateCluster.SpotProvisioningSpecificationProperty(
allocation_strategy=tasks.EmrCreateCluster.SpotAllocationStrategy.CAPACITY_OPTIMIZED,
timeout_action=tasks.EmrCreateCluster.SpotTimeoutAction.TERMINATE_CLUSTER,
timeout=Duration.minutes(5)
)
)
)]
),
name="SpotCluster",
integration_pattern=sfn.IntegrationPattern.RUN_JOB
)
java
EmrCreateCluster.Builder.create(this, "OnDemandSpecification")
.instances(InstancesConfigProperty.builder()
.instanceFleets(List.of(InstanceFleetConfigProperty.builder()
.instanceFleetType(EmrCreateCluster.getInstanceRoleType().MASTER)
.launchSpecifications(InstanceFleetProvisioningSpecificationsProperty.builder()
.onDemandSpecification(OnDemandProvisioningSpecificationProperty.builder()
.allocationStrategy(EmrCreateCluster.getOnDemandAllocationStrategy().LOWEST_PRICE)
.build())
.build())
.build()))
.build())
.name("OnDemandCluster")
.integrationPattern(IntegrationPattern.RUN_JOB)
.build();
EmrCreateCluster.Builder.create(this, "SpotSpecification")
.instances(InstancesConfigProperty.builder()
.instanceFleets(List.of(InstanceFleetConfigProperty.builder()
.instanceFleetType(EmrCreateCluster.getInstanceRoleType().MASTER)
.launchSpecifications(InstanceFleetProvisioningSpecificationsProperty.builder()
.spotSpecification(SpotProvisioningSpecificationProperty.builder()
.allocationStrategy(EmrCreateCluster.getSpotAllocationStrategy().CAPACITY_OPTIMIZED)
.timeoutAction(EmrCreateCluster.getSpotTimeoutAction().TERMINATE_CLUSTER)
.timeout(Duration.minutes(5))
.build())
.build())
.build()))
.build())
.name("SpotCluster")
.integrationPattern(IntegrationPattern.RUN_JOB)
.build();
csharp
new EmrCreateCluster(this, "OnDemandSpecification", new EmrCreateClusterProps {
Instances = new InstancesConfigProperty {
InstanceFleets = new [] { new InstanceFleetConfigProperty {
InstanceFleetType = EmrCreateCluster.InstanceRoleType.MASTER,
LaunchSpecifications = new InstanceFleetProvisioningSpecificationsProperty {
OnDemandSpecification = new OnDemandProvisioningSpecificationProperty {
AllocationStrategy = EmrCreateCluster.OnDemandAllocationStrategy.LOWEST_PRICE
}
}
} }
},
Name = "OnDemandCluster",
IntegrationPattern = IntegrationPattern.RUN_JOB
});
new EmrCreateCluster(this, "SpotSpecification", new EmrCreateClusterProps {
Instances = new InstancesConfigProperty {
InstanceFleets = new [] { new InstanceFleetConfigProperty {
InstanceFleetType = EmrCreateCluster.InstanceRoleType.MASTER,
LaunchSpecifications = new InstanceFleetProvisioningSpecificationsProperty {
SpotSpecification = new SpotProvisioningSpecificationProperty {
AllocationStrategy = EmrCreateCluster.SpotAllocationStrategy.CAPACITY_OPTIMIZED,
TimeoutAction = EmrCreateCluster.SpotTimeoutAction.TERMINATE_CLUSTER,
Timeout = Duration.Minutes(5)
}
}
} }
},
Name = "SpotCluster",
IntegrationPattern = IntegrationPattern.RUN_JOB
});
go
tasks.NewEmrCreateCluster(this, jsii.String("OnDemandSpecification"), &EmrCreateClusterProps{
Instances: &InstancesConfigProperty{
InstanceFleets: []InstanceFleetConfigProperty{
&InstanceFleetConfigProperty{
InstanceFleetType: tasks.EmrCreateCluster.InstanceRoleType_MASTER,
LaunchSpecifications: &InstanceFleetProvisioningSpecificationsProperty{
OnDemandSpecification: &OnDemandProvisioningSpecificationProperty{
AllocationStrategy: tasks.EmrCreateCluster.OnDemandAllocationStrategy_LOWEST_PRICE,
},
},
},
},
},
Name: jsii.String("OnDemandCluster"),
IntegrationPattern: sfn.IntegrationPattern_RUN_JOB,
})
tasks.NewEmrCreateCluster(this, jsii.String("SpotSpecification"), &EmrCreateClusterProps{
Instances: &InstancesConfigProperty{
InstanceFleets: []InstanceFleetConfigProperty{
&InstanceFleetConfigProperty{
InstanceFleetType: tasks.EmrCreateCluster.InstanceRoleType_MASTER,
LaunchSpecifications: &InstanceFleetProvisioningSpecificationsProperty{
SpotSpecification: &SpotProvisioningSpecificationProperty{
AllocationStrategy: tasks.EmrCreateCluster.SpotAllocationStrategy_CAPACITY_OPTIMIZED,
TimeoutAction: tasks.EmrCreateCluster.SpotTimeoutAction_TERMINATE_CLUSTER,
Timeout: awscdk.Duration_Minutes(jsii.Number(5)),
},
},
},
},
},
Name: jsii.String("SpotCluster"),
IntegrationPattern: sfn.IntegrationPattern_RUN_JOB,
})
You can use the prioritized allocation strategy to specify instance type priorities for On-Demand instances:
ts
new tasks.EmrCreateCluster(this, 'PrioritizedAllocation', {
instances: {
instanceFleets: [{
instanceFleetType: tasks.EmrCreateCluster.InstanceRoleType.CORE,
instanceTypeConfigs: [{
instanceType: 'm5.large',
priority: 0, // Highest priority
}, {
instanceType: 'm5.xlarge',
priority: 1, // Lower priority
}],
launchSpecifications: {
onDemandSpecification: {
allocationStrategy: tasks.EmrCreateCluster.OnDemandAllocationStrategy.PRIORITIZED,
},
},
targetOnDemandCapacity: 2,
}],
},
name: 'PrioritizedCluster',
integrationPattern: sfn.IntegrationPattern.RUN_JOB,
});
python
tasks.EmrCreateCluster(self, "PrioritizedAllocation",
instances=tasks.EmrCreateCluster.InstancesConfigProperty(
instance_fleets=[tasks.EmrCreateCluster.InstanceFleetConfigProperty(
instance_fleet_type=tasks.EmrCreateCluster.InstanceRoleType.CORE,
instance_type_configs=[tasks.EmrCreateCluster.InstanceTypeConfigProperty(
instance_type="m5.large",
priority=0
), tasks.EmrCreateCluster.InstanceTypeConfigProperty(
instance_type="m5.xlarge",
priority=1
)],
launch_specifications=tasks.EmrCreateCluster.InstanceFleetProvisioningSpecificationsProperty(
on_demand_specification=tasks.EmrCreateCluster.OnDemandProvisioningSpecificationProperty(
allocation_strategy=tasks.EmrCreateCluster.OnDemandAllocationStrategy.PRIORITIZED
)
),
target_on_demand_capacity=2
)]
),
name="PrioritizedCluster",
integration_pattern=sfn.IntegrationPattern.RUN_JOB
)
java
EmrCreateCluster.Builder.create(this, "PrioritizedAllocation")
.instances(InstancesConfigProperty.builder()
.instanceFleets(List.of(InstanceFleetConfigProperty.builder()
.instanceFleetType(EmrCreateCluster.getInstanceRoleType().CORE)
.instanceTypeConfigs(List.of(InstanceTypeConfigProperty.builder()
.instanceType("m5.large")
.priority(0)
.build(), InstanceTypeConfigProperty.builder()
.instanceType("m5.xlarge")
.priority(1)
.build()))
.launchSpecifications(InstanceFleetProvisioningSpecificationsProperty.builder()
.onDemandSpecification(OnDemandProvisioningSpecificationProperty.builder()
.allocationStrategy(EmrCreateCluster.getOnDemandAllocationStrategy().PRIORITIZED)
.build())
.build())
.targetOnDemandCapacity(2)
.build()))
.build())
.name("PrioritizedCluster")
.integrationPattern(IntegrationPattern.RUN_JOB)
.build();
csharp
new EmrCreateCluster(this, "PrioritizedAllocation", new EmrCreateClusterProps {
Instances = new InstancesConfigProperty {
InstanceFleets = new [] { new InstanceFleetConfigProperty {
InstanceFleetType = EmrCreateCluster.InstanceRoleType.CORE,
InstanceTypeConfigs = new [] { new InstanceTypeConfigProperty {
InstanceType = "m5.large",
Priority = 0
}, new InstanceTypeConfigProperty {
InstanceType = "m5.xlarge",
Priority = 1
} },
LaunchSpecifications = new InstanceFleetProvisioningSpecificationsProperty {
OnDemandSpecification = new OnDemandProvisioningSpecificationProperty {
AllocationStrategy = EmrCreateCluster.OnDemandAllocationStrategy.PRIORITIZED
}
},
TargetOnDemandCapacity = 2
} }
},
Name = "PrioritizedCluster",
IntegrationPattern = IntegrationPattern.RUN_JOB
});
go
tasks.NewEmrCreateCluster(this, jsii.String("PrioritizedAllocation"), &EmrCreateClusterProps{
Instances: &InstancesConfigProperty{
InstanceFleets: []InstanceFleetConfigProperty{
&InstanceFleetConfigProperty{
InstanceFleetType: tasks.EmrCreateCluster.InstanceRoleType_CORE,
InstanceTypeConfigs: []InstanceTypeConfigProperty{
&InstanceTypeConfigProperty{
InstanceType: jsii.String("m5.large"),
Priority: jsii.Number(0),
},
&InstanceTypeConfigProperty{
InstanceType: jsii.String("m5.xlarge"),
Priority: jsii.Number(1),
},
},
LaunchSpecifications: &InstanceFleetProvisioningSpecificationsProperty{
OnDemandSpecification: &OnDemandProvisioningSpecificationProperty{
AllocationStrategy: tasks.EmrCreateCluster.OnDemandAllocationStrategy_PRIORITIZED,
},
},
TargetOnDemandCapacity: jsii.Number(2),
},
},
},
Name: jsii.String("PrioritizedCluster"),
IntegrationPattern: sfn.IntegrationPattern_RUN_JOB,
})
You can customize EBS root device volume.
ts
new tasks.EmrCreateCluster(this, 'Create Cluster', {
instances: {},
name: 'ClusterName',
ebsRootVolumeIops: 4000,
ebsRootVolumeSize: Size.gibibytes(20),
ebsRootVolumeThroughput: 200,
});
python
tasks.EmrCreateCluster(self, "Create Cluster",
instances=tasks.EmrCreateCluster.InstancesConfigProperty(),
name="ClusterName",
ebs_root_volume_iops=4000,
ebs_root_volume_size=Size.gibibytes(20),
ebs_root_volume_throughput=200
)
java
EmrCreateCluster.Builder.create(this, "Create Cluster")
.instances(InstancesConfigProperty.builder().build())
.name("ClusterName")
.ebsRootVolumeIops(4000)
.ebsRootVolumeSize(Size.gibibytes(20))
.ebsRootVolumeThroughput(200)
.build();
csharp
new EmrCreateCluster(this, "Create Cluster", new EmrCreateClusterProps {
Instances = new InstancesConfigProperty { },
Name = "ClusterName",
EbsRootVolumeIops = 4000,
EbsRootVolumeSize = Size.Gibibytes(20),
EbsRootVolumeThroughput = 200
});
go
tasks.NewEmrCreateCluster(this, jsii.String("Create Cluster"), &EmrCreateClusterProps{
Instances: &InstancesConfigProperty{
},
Name: jsii.String("ClusterName"),
EbsRootVolumeIops: jsii.Number(4000),
EbsRootVolumeSize: awscdk.Size_Gibibytes(jsii.Number(20)),
EbsRootVolumeThroughput: jsii.Number(200),
})
If you want to run multiple steps in parallel, you can specify the stepConcurrencyLevel property. The concurrency range is between 1 and 256 inclusive, where the default concurrency of 1 means no step concurrency is allowed. stepConcurrencyLevel requires the EMR release label to be 5.28.0 or above.
ts
new tasks.EmrCreateCluster(this, 'Create Cluster', {
instances: {},
name: sfn.TaskInput.fromJsonPathAt('$.ClusterName').value,
stepConcurrencyLevel: 10,
});
python
tasks.EmrCreateCluster(self, "Create Cluster",
instances=tasks.EmrCreateCluster.InstancesConfigProperty(),
name=sfn.TaskInput.from_json_path_at("$.ClusterName").value,
step_concurrency_level=10
)
java
EmrCreateCluster.Builder.create(this, "Create Cluster")
.instances(InstancesConfigProperty.builder().build())
.name(TaskInput.fromJsonPathAt("$.ClusterName").getValue())
.stepConcurrencyLevel(10)
.build();
csharp
new EmrCreateCluster(this, "Create Cluster", new EmrCreateClusterProps {
Instances = new InstancesConfigProperty { },
Name = TaskInput.FromJsonPathAt("$.ClusterName").Value,
StepConcurrencyLevel = 10
});
go
tasks.NewEmrCreateCluster(this, jsii.String("Create Cluster"), &EmrCreateClusterProps{
Instances: &InstancesConfigProperty{
},
Name: sfn.TaskInput_FromJsonPathAt(jsii.String("$.ClusterName")).value,
StepConcurrencyLevel: jsii.Number(10),
})
If you want to use an auto-termination policy, you can specify the autoTerminationPolicyIdleTimeout 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).
ts
new tasks.EmrCreateCluster(this, 'Create Cluster', {
instances: {},
name: 'ClusterName',
autoTerminationPolicyIdleTimeout: Duration.seconds(100),
});
python
tasks.EmrCreateCluster(self, "Create Cluster",
instances=tasks.EmrCreateCluster.InstancesConfigProperty(),
name="ClusterName",
auto_termination_policy_idle_timeout=Duration.seconds(100)
)
java
EmrCreateCluster.Builder.create(this, "Create Cluster")
.instances(InstancesConfigProperty.builder().build())
.name("ClusterName")
.autoTerminationPolicyIdleTimeout(Duration.seconds(100))
.build();
csharp
new EmrCreateCluster(this, "Create Cluster", new EmrCreateClusterProps {
Instances = new InstancesConfigProperty { },
Name = "ClusterName",
AutoTerminationPolicyIdleTimeout = Duration.Seconds(100)
});
go
tasks.NewEmrCreateCluster(this, jsii.String("Create Cluster"), &EmrCreateClusterProps{
Instances: &InstancesConfigProperty{
},
Name: jsii.String("ClusterName"),
AutoTerminationPolicyIdleTimeout: awscdk.Duration_Seconds(jsii.Number(100)),
})
If you want to use managed scaling, you can specify the managedScalingPolicy property.
ts
new tasks.EmrCreateCluster(this, 'CreateCluster', {
instances: {
instanceFleets: [
{
instanceFleetType: tasks.EmrCreateCluster.InstanceRoleType.CORE,
instanceTypeConfigs: [
{
instanceType: 'm5.xlarge',
},
],
targetOnDemandCapacity: 1,
},
{
instanceFleetType: tasks.EmrCreateCluster.InstanceRoleType.MASTER,
instanceTypeConfigs: [
{
instanceType: 'm5.xlarge',
},
],
targetOnDemandCapacity: 1,
},
],
},
name: 'ClusterName',
releaseLabel: 'emr-7.9.0',
managedScalingPolicy: {
computeLimits: {
unitType: tasks.EmrCreateCluster.ComputeLimitsUnitType.INSTANCE_FLEET_UNITS,
maximumCapacityUnits: 4,
minimumCapacityUnits: 1,
maximumOnDemandCapacityUnits: 4,
maximumCoreCapacityUnits: 2,
},
},
});
python
tasks.EmrCreateCluster(self, "CreateCluster",
instances=tasks.EmrCreateCluster.InstancesConfigProperty(
instance_fleets=[tasks.EmrCreateCluster.InstanceFleetConfigProperty(
instance_fleet_type=tasks.EmrCreateCluster.InstanceRoleType.CORE,
instance_type_configs=[tasks.EmrCreateCluster.InstanceTypeConfigProperty(
instance_type="m5.xlarge"
)
],
target_on_demand_capacity=1
), tasks.EmrCreateCluster.InstanceFleetConfigProperty(
instance_fleet_type=tasks.EmrCreateCluster.InstanceRoleType.MASTER,
instance_type_configs=[tasks.EmrCreateCluster.InstanceTypeConfigProperty(
instance_type="m5.xlarge"
)
],
target_on_demand_capacity=1
)
]
),
name="ClusterName",
release_label="emr-7.9.0",
managed_scaling_policy=tasks.EmrCreateCluster.ManagedScalingPolicyProperty(
compute_limits=tasks.EmrCreateCluster.ManagedScalingComputeLimitsProperty(
unit_type=tasks.EmrCreateCluster.ComputeLimitsUnitType.INSTANCE_FLEET_UNITS,
maximum_capacity_units=4,
minimum_capacity_units=1,
maximum_on_demand_capacity_units=4,
maximum_core_capacity_units=2
)
)
)
java
EmrCreateCluster.Builder.create(this, "CreateCluster")
.instances(InstancesConfigProperty.builder()
.instanceFleets(List.of(InstanceFleetConfigProperty.builder()
.instanceFleetType(EmrCreateCluster.getInstanceRoleType().CORE)
.instanceTypeConfigs(List.of(InstanceTypeConfigProperty.builder()
.instanceType("m5.xlarge")
.build()))
.targetOnDemandCapacity(1)
.build(), InstanceFleetConfigProperty.builder()
.instanceFleetType(EmrCreateCluster.getInstanceRoleType().MASTER)
.instanceTypeConfigs(List.of(InstanceTypeConfigProperty.builder()
.instanceType("m5.xlarge")
.build()))
.targetOnDemandCapacity(1)
.build()))
.build())
.name("ClusterName")
.releaseLabel("emr-7.9.0")
.managedScalingPolicy(ManagedScalingPolicyProperty.builder()
.computeLimits(ManagedScalingComputeLimitsProperty.builder()
.unitType(EmrCreateCluster.getComputeLimitsUnitType().INSTANCE_FLEET_UNITS)
.maximumCapacityUnits(4)
.minimumCapacityUnits(1)
.maximumOnDemandCapacityUnits(4)
.maximumCoreCapacityUnits(2)
.build())
.build())
.build();
csharp
new EmrCreateCluster(this, "CreateCluster", new EmrCreateClusterProps {
Instances = new InstancesConfigProperty {
InstanceFleets = new [] { new InstanceFleetConfigProperty {
InstanceFleetType = EmrCreateCluster.InstanceRoleType.CORE,
InstanceTypeConfigs = new [] { new InstanceTypeConfigProperty {
InstanceType = "m5.xlarge"
} },
TargetOnDemandCapacity = 1
}, new InstanceFleetConfigProperty {
InstanceFleetType = EmrCreateCluster.InstanceRoleType.MASTER,
InstanceTypeConfigs = new [] { new InstanceTypeConfigProperty {
InstanceType = "m5.xlarge"
} },
TargetOnDemandCapacity = 1
} }
},
Name = "ClusterName",
ReleaseLabel = "emr-7.9.0",
ManagedScalingPolicy = new ManagedScalingPolicyProperty {
ComputeLimits = new ManagedScalingComputeLimitsProperty {
UnitType = EmrCreateCluster.ComputeLimitsUnitType.INSTANCE_FLEET_UNITS,
MaximumCapacityUnits = 4,
MinimumCapacityUnits = 1,
MaximumOnDemandCapacityUnits = 4,
MaximumCoreCapacityUnits = 2
}
}
});
go
tasks.NewEmrCreateCluster(this, jsii.String("CreateCluster"), &EmrCreateClusterProps{
Instances: &InstancesConfigProperty{
InstanceFleets: []InstanceFleetConfigProperty{
&InstanceFleetConfigProperty{
InstanceFleetType: tasks.EmrCreateCluster.InstanceRoleType_CORE,
InstanceTypeConfigs: []InstanceTypeConfigProperty{
&InstanceTypeConfigProperty{
InstanceType: jsii.String("m5.xlarge"),
},
},
TargetOnDemandCapacity: jsii.Number(1),
},
&InstanceFleetConfigProperty{
InstanceFleetType: tasks.EmrCreateCluster.InstanceRoleType_MASTER,
InstanceTypeConfigs: []InstanceTypeConfigProperty{
&InstanceTypeConfigProperty{
InstanceType: jsii.String("m5.xlarge"),
},
},
TargetOnDemandCapacity: jsii.Number(1),
},
},
},
Name: jsii.String("ClusterName"),
ReleaseLabel: jsii.String("emr-7.9.0"),
ManagedScalingPolicy: &ManagedScalingPolicyProperty{
ComputeLimits: &ManagedScalingComputeLimitsProperty{
UnitType: tasks.EmrCreateCluster.ComputeLimitsUnitType_INSTANCE_FLEET_UNITS,
MaximumCapacityUnits: jsii.Number(4),
MinimumCapacityUnits: jsii.Number(1),
MaximumOnDemandCapacityUnits: jsii.Number(4),
MaximumCoreCapacityUnits: jsii.Number(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 setTerminationProtection API in EMR.
ts
new tasks.EmrSetClusterTerminationProtection(this, 'Task', {
clusterId: 'ClusterId',
terminationProtected: false,
});
python
tasks.EmrSetClusterTerminationProtection(self, "Task",
cluster_id="ClusterId",
termination_protected=False
)
java
EmrSetClusterTerminationProtection.Builder.create(this, "Task")
.clusterId("ClusterId")
.terminationProtected(false)
.build();
csharp
new EmrSetClusterTerminationProtection(this, "Task", new EmrSetClusterTerminationProtectionProps {
ClusterId = "ClusterId",
TerminationProtected = false
});
go
tasks.NewEmrSetClusterTerminationProtection(this, jsii.String("Task"), &EmrSetClusterTerminationProtectionProps{
ClusterId: jsii.String("ClusterId"),
TerminationProtected: jsii.Boolean(false),
})
Terminate Cluster
Shuts down a cluster (job flow). Corresponds to the terminateJobFlows API in EMR.
ts
new tasks.EmrTerminateCluster(this, 'Task', {
clusterId: 'ClusterId',
});
python
tasks.EmrTerminateCluster(self, "Task",
cluster_id="ClusterId"
)
java
EmrTerminateCluster.Builder.create(this, "Task")
.clusterId("ClusterId")
.build();
csharp
new EmrTerminateCluster(this, "Task", new EmrTerminateClusterProps {
ClusterId = "ClusterId"
});
go
tasks.NewEmrTerminateCluster(this, jsii.String("Task"), &EmrTerminateClusterProps{
ClusterId: jsii.String("ClusterId"),
})
Add Step
Adds a new step to a running cluster. Corresponds to the addJobFlowSteps API in EMR.
ts
new tasks.EmrAddStep(this, 'Task', {
clusterId: 'ClusterId',
name: 'StepName',
jar: 'Jar',
actionOnFailure: tasks.ActionOnFailure.CONTINUE,
});
python
tasks.EmrAddStep(self, "Task",
cluster_id="ClusterId",
name="StepName",
jar="Jar",
action_on_failure=tasks.ActionOnFailure.CONTINUE
)
java
EmrAddStep.Builder.create(this, "Task")
.clusterId("ClusterId")
.name("StepName")
.jar("Jar")
.actionOnFailure(ActionOnFailure.CONTINUE)
.build();
csharp
new EmrAddStep(this, "Task", new EmrAddStepProps {
ClusterId = "ClusterId",
Name = "StepName",
Jar = "Jar",
ActionOnFailure = ActionOnFailure.CONTINUE
});
go
tasks.NewEmrAddStep(this, jsii.String("Task"), &EmrAddStepProps{
ClusterId: jsii.String("ClusterId"),
Name: jsii.String("StepName"),
Jar: jsii.String("Jar"),
ActionOnFailure: tasks.ActionOnFailure_CONTINUE,
})
To specify a custom runtime role use the executionRoleArn 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.
ts
import * as emr from 'aws-cdk-lib/aws-emr';
const cfnSecurityConfiguration = new emr.CfnSecurityConfiguration(this, 'EmrSecurityConfiguration', {
name: 'AddStepRuntimeRoleSecConfig',
securityConfiguration: JSON.parse(`
{
"AuthorizationConfiguration": {
"IAMConfiguration": {
"EnableApplicationScopedIAMRole": true,
"ApplicationScopedIAMRoleConfiguration":
{
"PropagateSourceIdentity": true
}
},
"LakeFormationConfiguration": {
"AuthorizedSessionTagValue": "Amazon EMR"
}
}
}`),
});
const task = new tasks.EmrCreateCluster(this, 'Create Cluster', {
instances: {},
name: sfn.TaskInput.fromJsonPathAt('$.ClusterName').value,
securityConfiguration: cfnSecurityConfiguration.name,
});
const executionRole = new iam.Role(this, 'Role', {
assumedBy: new iam.ArnPrincipal(task.clusterRole.roleArn),
});
executionRole.assumeRolePolicy?.addStatements(
new iam.PolicyStatement({
effect: iam.Effect.ALLOW,
principals: [
task.clusterRole,
],
actions: [
'sts:SetSourceIdentity',
],
}),
new iam.PolicyStatement({
effect: iam.Effect.ALLOW,
principals: [
task.clusterRole,
],
actions: [
'sts:TagSession',
],
conditions: {
StringEquals: {
'aws:RequestTag/LakeFormationAuthorizedCaller': 'Amazon EMR',
},
},
}),
);
new tasks.EmrAddStep(this, 'Task', {
clusterId: 'ClusterId',
executionRoleArn: executionRole.roleArn,
name: 'StepName',
jar: 'Jar',
actionOnFailure: tasks.ActionOnFailure.CONTINUE,
});
python
import aws_cdk.aws_emr as emr
cfn_security_configuration = emr.CfnSecurityConfiguration(self, "EmrSecurityConfiguration",
name="AddStepRuntimeRoleSecConfig",
security_configuration=JSON.parse("""
{
"AuthorizationConfiguration": {
"IAMConfiguration": {
"EnableApplicationScopedIAMRole": true,
"ApplicationScopedIAMRoleConfiguration":
{
"PropagateSourceIdentity": true
}
},
"LakeFormationConfiguration": {
"AuthorizedSessionTagValue": "Amazon EMR"
}
}
}""")
)
task = tasks.EmrCreateCluster(self, "Create Cluster",
instances=tasks.EmrCreateCluster.InstancesConfigProperty(),
name=sfn.TaskInput.from_json_path_at("$.ClusterName").value,
security_configuration=cfn_security_configuration.name
)
execution_role = iam.Role(self, "Role",
assumed_by=iam.ArnPrincipal(task.cluster_role.role_arn)
)
execution_role.assume_role_policy.add_statements(
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
principals=[task.cluster_role
],
actions=["sts:SetSourceIdentity"
]
),
iam.PolicyStatement(
effect=iam.Effect.ALLOW,
principals=[task.cluster_role
],
actions=["sts:TagSession"
],
conditions={
"StringEquals": {
"aws:RequestTag/LakeFormationAuthorizedCaller": "Amazon EMR"
}
}
))
tasks.EmrAddStep(self, "Task",
cluster_id="ClusterId",
execution_role_arn=execution_role.role_arn,
name="StepName",
jar="Jar",
action_on_failure=tasks.ActionOnFailure.CONTINUE
)
java
import software.amazon.awscdk.services.emr.*;
CfnSecurityConfiguration cfnSecurityConfiguration = CfnSecurityConfiguration.Builder.create(this, "EmrSecurityConfiguration")
.name("AddStepRuntimeRoleSecConfig")
.securityConfiguration(JSON.parse("\n {\n \"AuthorizationConfiguration\": {\n \"IAMConfiguration\": {\n \"EnableApplicationScopedIAMRole\": true,\n \"ApplicationScopedIAMRoleConfiguration\":\n {\n \"PropagateSourceIdentity\": true\n }\n },\n \"LakeFormationConfiguration\": {\n \"AuthorizedSessionTagValue\": \"Amazon EMR\"\n }\n }\n }"))
.build();
EmrCreateCluster task = EmrCreateCluster.Builder.create(this, "Create Cluster")
.instances(InstancesConfigProperty.builder().build())
.name(TaskInput.fromJsonPathAt("$.ClusterName").getValue())
.securityConfiguration(cfnSecurityConfiguration.getName())
.build();
Role executionRole = Role.Builder.create(this, "Role")
.assumedBy(new ArnPrincipal(task.getClusterRole().getRoleArn()))
.build();
executionRole.assumeRolePolicy.addStatements(
PolicyStatement.Builder.create()
.effect(Effect.ALLOW)
.principals(List.of(task.getClusterRole()))
.actions(List.of("sts:SetSourceIdentity"))
.build(),
PolicyStatement.Builder.create()
.effect(Effect.ALLOW)
.principals(List.of(task.getClusterRole()))
.actions(List.of("sts:TagSession"))
.conditions(Map.of(
"StringEquals", Map.of(
"aws:RequestTag/LakeFormationAuthorizedCaller", "Amazon EMR")))
.build());
EmrAddStep.Builder.create(this, "Task")
.clusterId("ClusterId")
.executionRoleArn(executionRole.getRoleArn())
.name("StepName")
.jar("Jar")
.actionOnFailure(ActionOnFailure.CONTINUE)
.build();
csharp
using Amazon.CDK.AWS.EMR;
var cfnSecurityConfiguration = new CfnSecurityConfiguration(this, "EmrSecurityConfiguration", new CfnSecurityConfigurationProps {
Name = "AddStepRuntimeRoleSecConfig",
SecurityConfiguration = JSON.Parse(@"
{
""AuthorizationConfiguration"": {
""IAMConfiguration"": {
""EnableApplicationScopedIAMRole"": true,
""ApplicationScopedIAMRoleConfiguration"":
{
""PropagateSourceIdentity"": true
}
},
""LakeFormationConfiguration"": {
""AuthorizedSessionTagValue"": ""Amazon EMR""
}
}
}")
});
var task = new EmrCreateCluster(this, "Create Cluster", new EmrCreateClusterProps {
Instances = new InstancesConfigProperty { },
Name = TaskInput.FromJsonPathAt("$.ClusterName").Value,
SecurityConfiguration = cfnSecurityConfiguration.Name
});
var executionRole = new Role(this, "Role", new RoleProps {
AssumedBy = new ArnPrincipal(task.ClusterRole.RoleArn)
});
executionRole.AssumeRolePolicy.AddStatements(
new PolicyStatement(new PolicyStatementProps {
Effect = Effect.ALLOW,
Principals = new [] { task.ClusterRole },
Actions = new [] { "sts:SetSourceIdentity" }
}),
new PolicyStatement(new PolicyStatementProps {
Effect = Effect.ALLOW,
Principals = new [] { task.ClusterRole },
Actions = new [] { "sts:TagSession" },
Conditions = new Dictionary<string, object> {
{ "StringEquals", new Dictionary<string, string> {
{ "aws:RequestTag/LakeFormationAuthorizedCaller", "Amazon EMR" }
} }
}
}));
new EmrAddStep(this, "Task", new EmrAddStepProps {
ClusterId = "ClusterId",
ExecutionRoleArn = executionRole.RoleArn,
Name = "StepName",
Jar = "Jar",
ActionOnFailure = ActionOnFailure.CONTINUE
});
go
import emr "github.com/aws/aws-cdk-go/awscdk"
cfnSecurityConfiguration := emr.NewCfnSecurityConfiguration(this, jsii.String("EmrSecurityConfiguration"), &CfnSecurityConfigurationProps{
Name: jsii.String("AddStepRuntimeRoleSecConfig"),
SecurityConfiguration: jSON.parse(jsii.String(`
{
"AuthorizationConfiguration": {
"IAMConfiguration": {
"EnableApplicationScopedIAMRole": true,
"ApplicationScopedIAMRoleConfiguration":
{
"PropagateSourceIdentity": true
}
},
"LakeFormationConfiguration": {
"AuthorizedSessionTagValue": "Amazon EMR"
}
}
}`)),
})
task := tasks.NewEmrCreateCluster(this, jsii.String("Create Cluster"), &EmrCreateClusterProps{
Instances: &InstancesConfigProperty{
},
Name: sfn.TaskInput_FromJsonPathAt(jsii.String("$.ClusterName")).value,
SecurityConfiguration: cfnSecurityConfiguration.name,
})
executionRole := iam.NewRole(this, jsii.String("Role"), &RoleProps{
AssumedBy: iam.NewArnPrincipal(task.clusterRole.RoleArn),
})
executionRole.AssumeRolePolicy.AddStatements(
iam.NewPolicyStatement(&PolicyStatementProps{
Effect: iam.Effect_ALLOW,
Principals: []IPrincipal{
task.clusterRole,
},
Actions: []*string{
jsii.String("sts:SetSourceIdentity"),
},
}),
iam.NewPolicyStatement(&PolicyStatementProps{
Effect: iam.Effect_ALLOW,
Principals: []IPrincipal{
task.clusterRole,
},
Actions: []*string{
jsii.String("sts:TagSession"),
},
Conditions: map[string]interface{}{
"StringEquals": map[string]*string{
"aws:RequestTag/LakeFormationAuthorizedCaller": jsii.String("Amazon EMR"),
},
},
}))
tasks.NewEmrAddStep(this, jsii.String("Task"), &EmrAddStepProps{
ClusterId: jsii.String("ClusterId"),
ExecutionRoleArn: executionRole.roleArn,
Name: jsii.String("StepName"),
Jar: jsii.String("Jar"),
ActionOnFailure: tasks.ActionOnFailure_CONTINUE,
})
Cancel Step
Cancels a pending step in a running cluster. Corresponds to the cancelSteps API in EMR.
ts
new tasks.EmrCancelStep(this, 'Task', {
clusterId: 'ClusterId',
stepId: 'StepId',
});
python
tasks.EmrCancelStep(self, "Task",
cluster_id="ClusterId",
step_id="StepId"
)
java
EmrCancelStep.Builder.create(this, "Task")
.clusterId("ClusterId")
.stepId("StepId")
.build();
csharp
new EmrCancelStep(this, "Task", new EmrCancelStepProps {
ClusterId = "ClusterId",
StepId = "StepId"
});
go
tasks.NewEmrCancelStep(this, jsii.String("Task"), &EmrCancelStepProps{
ClusterId: jsii.String("ClusterId"),
StepId: jsii.String("StepId"),
})
Modify Instance Fleet
Modifies the target On-Demand and target Spot capacities for the instance fleet with the specified InstanceFleetName.
Corresponds to the modifyInstanceFleet API in EMR.
ts
new tasks.EmrModifyInstanceFleetByName(this, 'Task', {
clusterId: 'ClusterId',
instanceFleetName: 'InstanceFleetName',
targetOnDemandCapacity: 2,
targetSpotCapacity: 0,
});
python
tasks.EmrModifyInstanceFleetByName(self, "Task",
cluster_id="ClusterId",
instance_fleet_name="InstanceFleetName",
target_on_demand_capacity=2,
target_spot_capacity=0
)
java
EmrModifyInstanceFleetByName.Builder.create(this, "Task")
.clusterId("ClusterId")
.instanceFleetName("InstanceFleetName")
.targetOnDemandCapacity(2)
.targetSpotCapacity(0)
.build();
csharp
new EmrModifyInstanceFleetByName(this, "Task", new EmrModifyInstanceFleetByNameProps {
ClusterId = "ClusterId",
InstanceFleetName = "InstanceFleetName",
TargetOnDemandCapacity = 2,
TargetSpotCapacity = 0
});
go
tasks.NewEmrModifyInstanceFleetByName(this, jsii.String("Task"), &EmrModifyInstanceFleetByNameProps{
ClusterId: jsii.String("ClusterId"),
InstanceFleetName: jsii.String("InstanceFleetName"),
TargetOnDemandCapacity: jsii.Number(2),
TargetSpotCapacity: jsii.Number(0),
})
Modify Instance Group
Modifies the number of nodes and configuration settings of an instance group.
Corresponds to the modifyInstanceGroups API in EMR.
ts
new tasks.EmrModifyInstanceGroupByName(this, 'Task', {
clusterId: 'ClusterId',
instanceGroupName: sfn.JsonPath.stringAt('$.InstanceGroupName'),
instanceGroup: {
instanceCount: 1,
},
});
python
tasks.EmrModifyInstanceGroupByName(self, "Task",
cluster_id="ClusterId",
instance_group_name=sfn.JsonPath.string_at("$.InstanceGroupName"),
instance_group=tasks.EmrModifyInstanceGroupByName.InstanceGroupModifyConfigProperty(
instance_count=1
)
)
java
EmrModifyInstanceGroupByName.Builder.create(this, "Task")
.clusterId("ClusterId")
.instanceGroupName(JsonPath.stringAt("$.InstanceGroupName"))
.instanceGroup(InstanceGroupModifyConfigProperty.builder()
.instanceCount(1)
.build())
.build();
csharp
new EmrModifyInstanceGroupByName(this, "Task", new EmrModifyInstanceGroupByNameProps {
ClusterId = "ClusterId",
InstanceGroupName = JsonPath.StringAt("$.InstanceGroupName"),
InstanceGroup = new InstanceGroupModifyConfigProperty {
InstanceCount = 1
}
});
go
tasks.NewEmrModifyInstanceGroupByName(this, jsii.String("Task"), &EmrModifyInstanceGroupByNameProps{
ClusterId: jsii.String("ClusterId"),
InstanceGroupName: sfn.JsonPath_StringAt(jsii.String("$.InstanceGroupName")),
InstanceGroup: &InstanceGroupModifyConfigProperty{
InstanceCount: jsii.Number(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.
ts
new tasks.EmrContainersCreateVirtualCluster(this, 'Create a Virtual Cluster', {
eksCluster: tasks.EksClusterInput.fromTaskInput(sfn.TaskInput.fromText('clusterId')),
});
python
tasks.EmrContainersCreateVirtualCluster(self, "Create a Virtual Cluster",
eks_cluster=tasks.EksClusterInput.from_task_input(sfn.TaskInput.from_text("clusterId"))
)
java
EmrContainersCreateVirtualCluster.Builder.create(this, "Create a Virtual Cluster")
.eksCluster(EksClusterInput.fromTaskInput(TaskInput.fromText("clusterId")))
.build();
csharp
new EmrContainersCreateVirtualCluster(this, "Create a Virtual Cluster", new EmrContainersCreateVirtualClusterProps {
EksCluster = EksClusterInput.FromTaskInput(TaskInput.FromText("clusterId"))
});
go
tasks.NewEmrContainersCreateVirtualCluster(this, jsii.String("Create a Virtual Cluster"), &EmrContainersCreateVirtualClusterProps{
EksCluster: tasks.EksClusterInput_FromTaskInput(sfn.TaskInput_FromText(jsii.String("clusterId"))),
})
The EKS cluster can also be passed in directly.
ts
import * as eks from 'aws-cdk-lib/aws-eks';
declare const eksCluster: eks.Cluster;
new tasks.EmrContainersCreateVirtualCluster(this, 'Create a Virtual Cluster', {
eksCluster: tasks.EksClusterInput.fromCluster(eksCluster),
});
python
import aws_cdk.aws_eks as eks
# eks_cluster: eks.Cluster
tasks.EmrContainersCreateVirtualCluster(self, "Create a Virtual Cluster",
eks_cluster=tasks.EksClusterInput.from_cluster(eks_cluster)
)
java
import software.amazon.awscdk.services.eks.*;
Cluster eksCluster;
EmrContainersCreateVirtualCluster.Builder.create(this, "Create a Virtual Cluster")
.eksCluster(EksClusterInput.fromCluster(eksCluster))
.build();
csharp
using Amazon.CDK.AWS.EKS;
Cluster eksCluster;
new EmrContainersCreateVirtualCluster(this, "Create a Virtual Cluster", new EmrContainersCreateVirtualClusterProps {
EksCluster = EksClusterInput.FromCluster(eksCluster)
});
go
import eks "github.com/aws/aws-cdk-go/awscdk"
var eksCluster Cluster
tasks.NewEmrContainersCreateVirtualCluster(this, jsii.String("Create a Virtual Cluster"), &EmrContainersCreateVirtualClusterProps{
EksCluster: tasks.EksClusterInput_FromCluster(eksCluster),
})
By default, the Kubernetes namespace that a virtual cluster maps to is "default", but a specific namespace within an EKS cluster can be selected.
ts
new tasks.EmrContainersCreateVirtualCluster(this, 'Create a Virtual Cluster', {
eksCluster: tasks.EksClusterInput.fromTaskInput(sfn.TaskInput.fromText('clusterId')),
eksNamespace: 'specified-namespace',
});
python
tasks.EmrContainersCreateVirtualCluster(self, "Create a Virtual Cluster",
eks_cluster=tasks.EksClusterInput.from_task_input(sfn.TaskInput.from_text("clusterId")),
eks_namespace="specified-namespace"
)
java
EmrContainersCreateVirtualCluster.Builder.create(this, "Create a Virtual Cluster")
.eksCluster(EksClusterInput.fromTaskInput(TaskInput.fromText("clusterId")))
.eksNamespace("specified-namespace")
.build();
csharp
new EmrContainersCreateVirtualCluster(this, "Create a Virtual Cluster", new EmrContainersCreateVirtualClusterProps {
EksCluster = EksClusterInput.FromTaskInput(TaskInput.FromText("clusterId")),
EksNamespace = "specified-namespace"
});
go
tasks.NewEmrContainersCreateVirtualCluster(this, jsii.String("Create a Virtual Cluster"), &EmrContainersCreateVirtualClusterProps{
EksCluster: tasks.EksClusterInput_FromTaskInput(sfn.TaskInput_FromText(jsii.String("clusterId"))),
EksNamespace: jsii.String("specified-namespace"),
})
Delete Virtual Cluster
The DeleteVirtualCluster API deletes a virtual cluster.
ts
new tasks.EmrContainersDeleteVirtualCluster(this, 'Delete a Virtual Cluster', {
virtualClusterId: sfn.TaskInput.fromJsonPathAt('$.virtualCluster'),
});
python
tasks.EmrContainersDeleteVirtualCluster(self, "Delete a Virtual Cluster",
virtual_cluster_id=sfn.TaskInput.from_json_path_at("$.virtualCluster")
)
java
EmrContainersDeleteVirtualCluster.Builder.create(this, "Delete a Virtual Cluster")
.virtualClusterId(TaskInput.fromJsonPathAt("$.virtualCluster"))
.build();
csharp
new EmrContainersDeleteVirtualCluster(this, "Delete a Virtual Cluster", new EmrContainersDeleteVirtualClusterProps {
VirtualClusterId = TaskInput.FromJsonPathAt("$.virtualCluster")
});
go
tasks.NewEmrContainersDeleteVirtualCluster(this, jsii.String("Delete a Virtual Cluster"), &EmrContainersDeleteVirtualClusterProps{
VirtualClusterId: sfn.TaskInput_FromJsonPathAt(jsii.String("$.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:
ts
new tasks.EmrContainersStartJobRun(this, 'EMR Containers Start Job Run', {
virtualCluster: tasks.VirtualClusterInput.fromVirtualClusterId('de92jdei2910fwedz'),
releaseLabel: tasks.ReleaseLabel.EMR_6_2_0,
jobDriver: {
sparkSubmitJobDriver: {
entryPoint: sfn.TaskInput.fromText('local:///usr/lib/spark/examples/src/main/python/pi.py'),
sparkSubmitParameters: '--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1',
},
},
});
python
tasks.EmrContainersStartJobRun(self, "EMR Containers Start Job Run",
virtual_cluster=tasks.VirtualClusterInput.from_virtual_cluster_id("de92jdei2910fwedz"),
release_label=tasks.ReleaseLabel.EMR_6_2_0,
job_driver=tasks.JobDriver(
spark_submit_job_driver=tasks.SparkSubmitJobDriver(
entry_point=sfn.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"
)
)
)
java
EmrContainersStartJobRun.Builder.create(this, "EMR Containers Start Job Run")
.virtualCluster(VirtualClusterInput.fromVirtualClusterId("de92jdei2910fwedz"))
.releaseLabel(ReleaseLabel.EMR_6_2_0)
.jobDriver(JobDriver.builder()
.sparkSubmitJobDriver(SparkSubmitJobDriver.builder()
.entryPoint(TaskInput.fromText("local:///usr/lib/spark/examples/src/main/python/pi.py"))
.sparkSubmitParameters("--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1")
.build())
.build())
.build();
csharp
new EmrContainersStartJobRun(this, "EMR Containers Start Job Run", new EmrContainersStartJobRunProps {
VirtualCluster = VirtualClusterInput.FromVirtualClusterId("de92jdei2910fwedz"),
ReleaseLabel = ReleaseLabel.EMR_6_2_0,
JobDriver = new JobDriver {
SparkSubmitJobDriver = new SparkSubmitJobDriver {
EntryPoint = TaskInput.FromText("local:///usr/lib/spark/examples/src/main/python/pi.py"),
SparkSubmitParameters = "--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1"
}
}
});
go
tasks.NewEmrContainersStartJobRun(this, jsii.String("EMR Containers Start Job Run"), &EmrContainersStartJobRunProps{
VirtualCluster: tasks.VirtualClusterInput_FromVirtualClusterId(jsii.String("de92jdei2910fwedz")),
ReleaseLabel: tasks.ReleaseLabel_EMR_6_2_0(),
JobDriver: &JobDriver{
SparkSubmitJobDriver: &SparkSubmitJobDriver{
EntryPoint: sfn.TaskInput_FromText(jsii.String("local:///usr/lib/spark/examples/src/main/python/pi.py")),
SparkSubmitParameters: jsii.String("--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:
ts
new tasks.EmrContainersStartJobRun(this, 'EMR Containers Start Job Run', {
virtualCluster: tasks.VirtualClusterInput.fromVirtualClusterId('de92jdei2910fwedz'),
releaseLabel: tasks.ReleaseLabel.EMR_6_2_0,
jobName: 'EMR-Containers-Job',
jobDriver: {
sparkSubmitJobDriver: {
entryPoint: sfn.TaskInput.fromText('local:///usr/lib/spark/examples/src/main/python/pi.py'),
},
},
applicationConfig: [{
classification: tasks.Classification.SPARK_DEFAULTS,
properties: {
'spark.executor.instances': '1',
'spark.executor.memory': '512M',
},
}],
});
python
tasks.EmrContainersStartJobRun(self, "EMR Containers Start Job Run",
virtual_cluster=tasks.VirtualClusterInput.from_virtual_cluster_id("de92jdei2910fwedz"),
release_label=tasks.ReleaseLabel.EMR_6_2_0,
job_name="EMR-Containers-Job",
job_driver=tasks.JobDriver(
spark_submit_job_driver=tasks.SparkSubmitJobDriver(
entry_point=sfn.TaskInput.from_text("local:///usr/lib/spark/examples/src/main/python/pi.py")
)
),
application_config=[tasks.ApplicationConfiguration(
classification=tasks.Classification.SPARK_DEFAULTS,
properties={
"spark.executor.instances": "1",
"spark.executor.memory": "512M"
}
)]
)
java
EmrContainersStartJobRun.Builder.create(this, "EMR Containers Start Job Run")
.virtualCluster(VirtualClusterInput.fromVirtualClusterId("de92jdei2910fwedz"))
.releaseLabel(ReleaseLabel.EMR_6_2_0)
.jobName("EMR-Containers-Job")
.jobDriver(JobDriver.builder()
.sparkSubmitJobDriver(SparkSubmitJobDriver.builder()
.entryPoint(TaskInput.fromText("local:///usr/lib/spark/examples/src/main/python/pi.py"))
.build())
.build())
.applicationConfig(List.of(ApplicationConfiguration.builder()
.classification(Classification.SPARK_DEFAULTS)
.properties(Map.of(
"spark.executor.instances", "1",
"spark.executor.memory", "512M"))
.build()))
.build();
csharp
new EmrContainersStartJobRun(this, "EMR Containers Start Job Run", new EmrContainersStartJobRunProps {
VirtualCluster = VirtualClusterInput.FromVirtualClusterId("de92jdei2910fwedz"),
ReleaseLabel = ReleaseLabel.EMR_6_2_0,
JobName = "EMR-Containers-Job",
JobDriver = new JobDriver {
SparkSubmitJobDriver = new SparkSubmitJobDriver {
EntryPoint = TaskInput.FromText("local:///usr/lib/spark/examples/src/main/python/pi.py")
}
},
ApplicationConfig = new [] { new ApplicationConfiguration {
Classification = Classification.SPARK_DEFAULTS,
Properties = new Dictionary<string, string> {
{ "spark.executor.instances", "1" },
{ "spark.executor.memory", "512M" }
}
} }
});
go
tasks.NewEmrContainersStartJobRun(this, jsii.String("EMR Containers Start Job Run"), &EmrContainersStartJobRunProps{
VirtualCluster: tasks.VirtualClusterInput_FromVirtualClusterId(jsii.String("de92jdei2910fwedz")),
ReleaseLabel: tasks.ReleaseLabel_EMR_6_2_0(),
JobName: jsii.String("EMR-Containers-Job"),
JobDriver: &JobDriver{
SparkSubmitJobDriver: &SparkSubmitJobDriver{
EntryPoint: sfn.TaskInput_FromText(jsii.String("local:///usr/lib/spark/examples/src/main/python/pi.py")),
},
},
ApplicationConfig: []ApplicationConfiguration{
&ApplicationConfiguration{
Classification: tasks.Classification_SPARK_DEFAULTS(),
Properties: map[string]*string{
"spark.executor.instances": jsii.String("1"),
"spark.executor.memory": jsii.String("512M"),
},
},
},
})
Job monitoring can be enabled if monitoring.logging is set true. This automatically generates an S3 bucket and CloudWatch logs.
ts
new tasks.EmrContainersStartJobRun(this, 'EMR Containers Start Job Run', {
virtualCluster: tasks.VirtualClusterInput.fromVirtualClusterId('de92jdei2910fwedz'),
releaseLabel: tasks.ReleaseLabel.EMR_6_2_0,
jobDriver: {
sparkSubmitJobDriver: {
entryPoint: sfn.TaskInput.fromText('local:///usr/lib/spark/examples/src/main/python/pi.py'),
sparkSubmitParameters: '--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1',
},
},
monitoring: {
logging: true,
},
});
python
tasks.EmrContainersStartJobRun(self, "EMR Containers Start Job Run",
virtual_cluster=tasks.VirtualClusterInput.from_virtual_cluster_id("de92jdei2910fwedz"),
release_label=tasks.ReleaseLabel.EMR_6_2_0,
job_driver=tasks.JobDriver(
spark_submit_job_driver=tasks.SparkSubmitJobDriver(
entry_point=sfn.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=tasks.Monitoring(
logging=True
)
)
java
EmrContainersStartJobRun.Builder.create(this, "EMR Containers Start Job Run")
.virtualCluster(VirtualClusterInput.fromVirtualClusterId("de92jdei2910fwedz"))
.releaseLabel(ReleaseLabel.EMR_6_2_0)
.jobDriver(JobDriver.builder()
.sparkSubmitJobDriver(SparkSubmitJobDriver.builder()
.entryPoint(TaskInput.fromText("local:///usr/lib/spark/examples/src/main/python/pi.py"))
.sparkSubmitParameters("--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1")
.build())
.build())
.monitoring(Monitoring.builder()
.logging(true)
.build())
.build();
csharp
new EmrContainersStartJobRun(this, "EMR Containers Start Job Run", new EmrContainersStartJobRunProps {
VirtualCluster = VirtualClusterInput.FromVirtualClusterId("de92jdei2910fwedz"),
ReleaseLabel = ReleaseLabel.EMR_6_2_0,
JobDriver = new JobDriver {
SparkSubmitJobDriver = new SparkSubmitJobDriver {
EntryPoint = TaskInput.FromText("local:///usr/lib/spark/examples/src/main/python/pi.py"),
SparkSubmitParameters = "--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1"
}
},
Monitoring = new Monitoring {
Logging = true
}
});
go
tasks.NewEmrContainersStartJobRun(this, jsii.String("EMR Containers Start Job Run"), &EmrContainersStartJobRunProps{
VirtualCluster: tasks.VirtualClusterInput_FromVirtualClusterId(jsii.String("de92jdei2910fwedz")),
ReleaseLabel: tasks.ReleaseLabel_EMR_6_2_0(),
JobDriver: &JobDriver{
SparkSubmitJobDriver: &SparkSubmitJobDriver{
EntryPoint: sfn.TaskInput_FromText(jsii.String("local:///usr/lib/spark/examples/src/main/python/pi.py")),
SparkSubmitParameters: jsii.String("--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1"),
},
},
Monitoring: &Monitoring{
Logging: jsii.Boolean(true),
},
})
Otherwise, providing monitoring for jobs with existing log groups and log buckets is also available.
ts
import * as logs from 'aws-cdk-lib/aws-logs';
const logGroup = new logs.LogGroup(this, 'Log Group');
const logBucket = new s3.Bucket(this, 'S3 Bucket')
new tasks.EmrContainersStartJobRun(this, 'EMR Containers Start Job Run', {
virtualCluster: tasks.VirtualClusterInput.fromVirtualClusterId('de92jdei2910fwedz'),
releaseLabel: tasks.ReleaseLabel.EMR_6_2_0,
jobDriver: {
sparkSubmitJobDriver: {
entryPoint: sfn.TaskInput.fromText('local:///usr/lib/spark/examples/src/main/python/pi.py'),
sparkSubmitParameters: '--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1',
},
},
monitoring: {
logGroup: logGroup,
logBucket: logBucket,
},
});
python
import aws_cdk.aws_logs as logs
log_group = logs.LogGroup(self, "Log Group")
log_bucket = s3.Bucket(self, "S3 Bucket")
tasks.EmrContainersStartJobRun(self, "EMR Containers Start Job Run",
virtual_cluster=tasks.VirtualClusterInput.from_virtual_cluster_id("de92jdei2910fwedz"),
release_label=tasks.ReleaseLabel.EMR_6_2_0,
job_driver=tasks.JobDriver(
spark_submit_job_driver=tasks.SparkSubmitJobDriver(
entry_point=sfn.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=tasks.Monitoring(
log_group=log_group,
log_bucket=log_bucket
)
)
java
import software.amazon.awscdk.services.logs.*;
LogGroup logGroup = new LogGroup(this, "Log Group");
Bucket logBucket = new Bucket(this, "S3 Bucket");
EmrContainersStartJobRun.Builder.create(this, "EMR Containers Start Job Run")
.virtualCluster(VirtualClusterInput.fromVirtualClusterId("de92jdei2910fwedz"))
.releaseLabel(ReleaseLabel.EMR_6_2_0)
.jobDriver(JobDriver.builder()
.sparkSubmitJobDriver(SparkSubmitJobDriver.builder()
.entryPoint(TaskInput.fromText("local:///usr/lib/spark/examples/src/main/python/pi.py"))
.sparkSubmitParameters("--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1")
.build())
.build())
.monitoring(Monitoring.builder()
.logGroup(logGroup)
.logBucket(logBucket)
.build())
.build();
csharp
using Amazon.CDK.AWS.Logs;
var logGroup = new LogGroup(this, "Log Group");
var logBucket = new Bucket(this, "S3 Bucket");
new EmrContainersStartJobRun(this, "EMR Containers Start Job Run", new EmrContainersStartJobRunProps {
VirtualCluster = VirtualClusterInput.FromVirtualClusterId("de92jdei2910fwedz"),
ReleaseLabel = ReleaseLabel.EMR_6_2_0,
JobDriver = new JobDriver {
SparkSubmitJobDriver = new SparkSubmitJobDriver {
EntryPoint = TaskInput.FromText("local:///usr/lib/spark/examples/src/main/python/pi.py"),
SparkSubmitParameters = "--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1"
}
},
Monitoring = new Monitoring {
LogGroup = logGroup,
LogBucket = logBucket
}
});
go
import logs "github.com/aws/aws-cdk-go/awscdk"
logGroup := logs.NewLogGroup(this, jsii.String("Log Group"))
logBucket := s3.NewBucket(this, jsii.String("S3 Bucket"))
tasks.NewEmrContainersStartJobRun(this, jsii.String("EMR Containers Start Job Run"), &EmrContainersStartJobRunProps{
VirtualCluster: tasks.VirtualClusterInput_FromVirtualClusterId(jsii.String("de92jdei2910fwedz")),
ReleaseLabel: tasks.ReleaseLabel_EMR_6_2_0(),
JobDriver: &JobDriver{
SparkSubmitJobDriver: &SparkSubmitJobDriver{
EntryPoint: sfn.TaskInput_FromText(jsii.String("local:///usr/lib/spark/examples/src/main/python/pi.py")),
SparkSubmitParameters: jsii.String("--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1"),
},
},
Monitoring: &Monitoring{
LogGroup: logGroup,
LogBucket: logBucket,
},
})
Users can provide their own existing Job Execution Role.
ts
new tasks.EmrContainersStartJobRun(this, 'EMR Containers Start Job Run', {
virtualCluster:tasks.VirtualClusterInput.fromTaskInput(sfn.TaskInput.fromJsonPathAt('$.VirtualClusterId')),
releaseLabel: tasks.ReleaseLabel.EMR_6_2_0,
jobName: 'EMR-Containers-Job',
executionRole: iam.Role.fromRoleArn(this, 'Job-Execution-Role', 'arn:aws:iam::xxxxxxxxxxxx:role/JobExecutionRole'),
jobDriver: {
sparkSubmitJobDriver: {
entryPoint: sfn.TaskInput.fromText('local:///usr/lib/spark/examples/src/main/python/pi.py'),
sparkSubmitParameters: '--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1',
},
},
});
python
tasks.EmrContainersStartJobRun(self, "EMR Containers Start Job Run",
virtual_cluster=tasks.VirtualClusterInput.from_task_input(sfn.TaskInput.from_json_path_at("$.VirtualClusterId")),
release_label=tasks.ReleaseLabel.EMR_6_2_0,
job_name="EMR-Containers-Job",
execution_role=iam.Role.from_role_arn(self, "Job-Execution-Role", "arn:aws:iam::xxxxxxxxxxxx:role/JobExecutionRole"),
job_driver=tasks.JobDriver(
spark_submit_job_driver=tasks.SparkSubmitJobDriver(
entry_point=sfn.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"
)
)
)
java
EmrContainersStartJobRun.Builder.create(this, "EMR Containers Start Job Run")
.virtualCluster(VirtualClusterInput.fromTaskInput(TaskInput.fromJsonPathAt("$.VirtualClusterId")))
.releaseLabel(ReleaseLabel.EMR_6_2_0)
.jobName("EMR-Containers-Job")
.executionRole(Role.fromRoleArn(this, "Job-Execution-Role", "arn:aws:iam::xxxxxxxxxxxx:role/JobExecutionRole"))
.jobDriver(JobDriver.builder()
.sparkSubmitJobDriver(SparkSubmitJobDriver.builder()
.entryPoint(TaskInput.fromText("local:///usr/lib/spark/examples/src/main/python/pi.py"))
.sparkSubmitParameters("--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1")
.build())
.build())
.build();
csharp
new EmrContainersStartJobRun(this, "EMR Containers Start Job Run", new EmrContainersStartJobRunProps {
VirtualCluster = VirtualClusterInput.FromTaskInput(TaskInput.FromJsonPathAt("$.VirtualClusterId")),
ReleaseLabel = ReleaseLabel.EMR_6_2_0,
JobName = "EMR-Containers-Job",
ExecutionRole = Role.FromRoleArn(this, "Job-Execution-Role", "arn:aws:iam::xxxxxxxxxxxx:role/JobExecutionRole"),
JobDriver = new JobDriver {
SparkSubmitJobDriver = new SparkSubmitJobDriver {
EntryPoint = TaskInput.FromText("local:///usr/lib/spark/examples/src/main/python/pi.py"),
SparkSubmitParameters = "--conf spark.executor.instances=2 --conf spark.executor.memory=2G --conf spark.executor.cores=2 --conf spark.driver.cores=1"
}
}
});
go
tasks.NewEmrContainersStartJobRun(this, jsii.String("EMR Containers Start Job Run"), &EmrContainersStartJobRunProps{
VirtualCluster: tasks.VirtualClusterInput_FromTaskInput(sfn.TaskInput_FromJsonPathAt(jsii.String("$.VirtualClusterId"))),
ReleaseLabel: tasks.ReleaseLabel_EMR_6_2_0(),
JobName: jsii.String("EMR-Containers-Job"),
ExecutionRole: iam.Role_FromRoleArn(this, jsii.String("Job-Execution-Role"), jsii.String("arn:aws:iam::xxxxxxxxxxxx:role/JobExecutionRole")),
JobDriver: &JobDriver{
SparkSubmitJobDriver: &SparkSubmitJobDriver{
EntryPoint: sfn.TaskInput_FromText(jsii.String("local:///usr/lib/spark/examples/src/main/python/pi.py")),
SparkSubmitParameters: jsii.String("--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.
ts
import * as eks from 'aws-cdk-lib/aws-eks';
import { KubectlV35Layer } from '@aws-cdk/lambda-layer-kubectl-v35';
const myEksCluster = new eks.Cluster(this, 'my sample cluster', {
version: eks.KubernetesVersion.V1_35,
clusterName: 'myEksCluster',
kubectlLayer: new KubectlV35Layer(this, 'kubectl'),
});
new tasks.EksCall(this, 'Call a EKS Endpoint', {
cluster: myEksCluster,
httpMethod: tasks.HttpMethods.GET,
httpPath: '/api/v1/namespaces/default/pods',
});
python
import aws_cdk.aws_eks as eks
from aws_cdk.lambda_layer_kubectl_v35 import KubectlV35Layer
my_eks_cluster = eks.Cluster(self, "my sample cluster",
version=eks.KubernetesVersion.V1_35,
cluster_name="myEksCluster",
kubectl_layer=KubectlV35Layer(self, "kubectl")
)
tasks.EksCall(self, "Call a EKS Endpoint",
cluster=my_eks_cluster,
http_method=tasks.HttpMethods.GET,
http_path="/api/v1/namespaces/default/pods"
)
java
import software.amazon.awscdk.services.eks.*;
import software.amazon.awscdk.cdk.lambdalayer.kubectl.v35.KubectlV35Layer;
Cluster myEksCluster = Cluster.Builder.create(this, "my sample cluster")
.version(KubernetesVersion.V1_35)
.clusterName("myEksCluster")
.kubectlLayer(new KubectlV35Layer(this, "kubectl"))
.build();
EksCall.Builder.create(this, "Call a EKS Endpoint")
.cluster(myEksCluster)
.httpMethod(HttpMethods.GET)
.httpPath("/api/v1/namespaces/default/pods")
.build();
csharp
using Amazon.CDK.AWS.EKS;
using Amazon.CDK.LambdaLayer.KubectlV35;
var myEksCluster = new Cluster(this, "my sample cluster", new ClusterProps {
Version = KubernetesVersion.V1_35,
ClusterName = "myEksCluster",
KubectlLayer = new KubectlV35Layer(this, "kubectl")
});
new EksCall(this, "Call a EKS Endpoint", new EksCallProps {
Cluster = myEksCluster,
HttpMethod = HttpMethods.GET,
HttpPath = "/api/v1/namespaces/default/pods"
});
go
import "github.com/aws/aws-cdk-go/awscdk"
import "github.com/cdklabs/awscdk-kubectl-go/kubectlv35"
myEksCluster := eks.NewCluster(this, jsii.String("my sample cluster"), &ClusterProps{
Version: eks.KubernetesVersion_V1_35(),
ClusterName: jsii.String("myEksCluster"),
KubectlLayer: kubectlv35.NewKubectlV35Layer(this, jsii.String("kubectl")),
})
tasks.NewEksCall(this, jsii.String("Call a EKS Endpoint"), &EksCallProps{
Cluster: myEksCluster,
HttpMethod: tasks.HttpMethods_GET,
HttpPath: jsii.String("/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.
ts
import * as events from 'aws-cdk-lib/aws-events';
const myEventBus = new events.EventBus(this, 'EventBus', {
eventBusName: 'MyEventBus1',
});
new tasks.EventBridgePutEvents(this, 'Send an event to EventBridge', {
entries: [{
detail: sfn.TaskInput.fromObject({
Message: 'Hello from Step Functions!',
}),
eventBus: myEventBus,
detailType: 'MessageFromStepFunctions',
source: 'step.functions',
}],
});
python
import aws_cdk.aws_events as events
my_event_bus = events.EventBus(self, "EventBus",
event_bus_name="MyEventBus1"
)
tasks.EventBridgePutEvents(self, "Send an event to EventBridge",
entries=[tasks.EventBridgePutEventsEntry(
detail=sfn.TaskInput.from_object({
"Message": "Hello from Step Functions!"
}),
event_bus=my_event_bus,
detail_type="MessageFromStepFunctions",
source="step.functions"
)]
)
java
import software.amazon.awscdk.services.events.*;
EventBus myEventBus = EventBus.Builder.create(this, "EventBus")
.eventBusName("MyEventBus1")
.build();
EventBridgePutEvents.Builder.create(this, "Send an event to EventBridge")
.entries(List.of(EventBridgePutEventsEntry.builder()
.detail(TaskInput.fromObject(Map.of(
"Message", "Hello from Step Functions!")))
.eventBus(myEventBus)
.detailType("MessageFromStepFunctions")
.source("step.functions")
.build()))
.build();
csharp
using Amazon.CDK.AWS.Events;
var myEventBus = new EventBus(this, "EventBus", new EventBusProps {
EventBusName = "MyEventBus1"
});
new EventBridgePutEvents(this, "Send an event to EventBridge", new EventBridgePutEventsProps {
Entries = new [] { new EventBridgePutEventsEntry {
Detail = TaskInput.FromObject(new Dictionary<string, object> {
{ "Message", "Hello from Step Functions!" }
}),
EventBus = myEventBus,
DetailType = "MessageFromStepFunctions",
Source = "step.functions"
} }
});
go
import events "github.com/aws/aws-cdk-go/awscdk"
myEventBus := events.NewEventBus(this, jsii.String("EventBus"), &EventBusProps{
EventBusName: jsii.String("MyEventBus1"),
})
tasks.NewEventBridgePutEvents(this, jsii.String("Send an event to EventBridge"), &EventBridgePutEventsProps{
Entries: []EventBridgePutEventsEntry{
&EventBridgePutEventsEntry{
Detail: sfn.TaskInput_FromObject(map[string]interface{}{
"Message": jsii.String("Hello from Step Functions!"),
}),
EventBus: myEventBus,
DetailType: jsii.String("MessageFromStepFunctions"),
Source: jsii.String("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:
ts
import * as scheduler from 'aws-cdk-lib/aws-scheduler';
import * as kms from 'aws-cdk-lib/aws-kms';
declare const key: kms.Key;
declare const scheduleGroup: scheduler.CfnScheduleGroup;
declare const targetQueue: sqs.Queue;
declare const deadLetterQueue: sqs.Queue;
const schedulerRole = new iam.Role(this, 'SchedulerRole', {
assumedBy: new iam.ServicePrincipal('scheduler.amazonaws.com'),
});
// To send the message to the queue
// This policy changes depending on the type of target.
schedulerRole.addToPrincipalPolicy(new iam.PolicyStatement({
actions: ['sqs:SendMessage'],
resources: [targetQueue.queueArn],
}));
const createScheduleTask1 = new tasks.EventBridgeSchedulerCreateScheduleTask(this, 'createSchedule', {
scheduleName: 'TestSchedule',
actionAfterCompletion: tasks.ActionAfterCompletion.NONE,
clientToken: 'testToken',
description: 'TestDescription',
startDate: new Date(),
endDate: new Date(new Date().getTime() + 1000 * 60 * 60),
flexibleTimeWindow: Duration.minutes(5),
groupName: scheduleGroup.ref,
kmsKey: key,
schedule: tasks.Schedule.rate(Duration.minutes(5)),
timezone: 'UTC',
enabled: true,
target: new tasks.EventBridgeSchedulerTarget({
arn: targetQueue.queueArn,
role: schedulerRole,
retryPolicy: {
maximumRetryAttempts: 2,
maximumEventAge: Duration.minutes(5),
},
deadLetterQueue,
}),
});
python
import aws_cdk.aws_scheduler as scheduler
import aws_cdk.aws_kms as kms
# key: kms.Key
# schedule_group: scheduler.CfnScheduleGroup
# target_queue: sqs.Queue
# dead_letter_queue: sqs.Queue
scheduler_role = iam.Role(self, "SchedulerRole",
assumed_by=iam.ServicePrincipal("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(iam.PolicyStatement(
actions=["sqs:SendMessage"],
resources=[target_queue.queue_arn]
))
create_schedule_task1 = tasks.EventBridgeSchedulerCreateScheduleTask(self, "createSchedule",
schedule_name="TestSchedule",
action_after_completion=tasks.ActionAfterCompletion.NONE,
client_token="testToken",
description="TestDescription",
start_date=Date(),
end_date=Date(Date().get_time() + 1000 * 60 * 60),
flexible_time_window=Duration.minutes(5),
group_name=schedule_group.ref,
kms_key=key,
schedule=tasks.Schedule.rate(Duration.minutes(5)),
timezone="UTC",
enabled=True,
target=tasks.EventBridgeSchedulerTarget(
arn=target_queue.queue_arn,
role=scheduler_role,
retry_policy=tasks.RetryPolicy(
maximum_retry_attempts=2,
maximum_event_age=Duration.minutes(5)
),
dead_letter_queue=dead_letter_queue
)
)
java
import software.amazon.awscdk.services.scheduler.*;
import software.amazon.awscdk.services.kms.*;
Key key;
CfnScheduleGroup scheduleGroup;
Queue targetQueue;
Queue deadLetterQueue;
Role schedulerRole = Role.Builder.create(this, "SchedulerRole")
.assumedBy(new ServicePrincipal("scheduler.amazonaws.com"))
.build();
// To send the message to the queue
// This policy changes depending on the type of target.
schedulerRole.addToPrincipalPolicy(PolicyStatement.Builder.create()
.actions(List.of("sqs:SendMessage"))
.resources(List.of(targetQueue.getQueueArn()))
.build());
EventBridgeSchedulerCreateScheduleTask createScheduleTask1 = EventBridgeSchedulerCreateScheduleTask.Builder.create(this, "createSchedule")
.scheduleName("TestSchedule")
.actionAfterCompletion(ActionAfterCompletion.NONE)
.clientToken("testToken")
.description("TestDescription")
.startDate(new Date())
.endDate(new Date(new Date().getTime() + 1000 * 60 * 60))
.flexibleTimeWindow(Duration.minutes(5))
.groupName(scheduleGroup.getRef())
.kmsKey(key)
.schedule(Schedule.rate(Duration.minutes(5)))
.timezone("UTC")
.enabled(true)
.target(EventBridgeSchedulerTarget.Builder.create()
.arn(targetQueue.getQueueArn())
.role(schedulerRole)
.retryPolicy(RetryPolicy.builder()
.maximumRetryAttempts(2)
.maximumEventAge(Duration.minutes(5))
.build())
.deadLetterQueue(deadLetterQueue)
.build())
.build();
csharp
using Amazon.CDK.AWS.Scheduler;
using Amazon.CDK.AWS.KMS;
Key key;
CfnScheduleGroup scheduleGroup;
Queue targetQueue;
Queue deadLetterQueue;
var schedulerRole = new Role(this, "SchedulerRole", new RoleProps {
AssumedBy = new ServicePrincipal("scheduler.amazonaws.com")
});
// To send the message to the queue
// This policy changes depending on the type of target.
schedulerRole.AddToPrincipalPolicy(new PolicyStatement(new PolicyStatementProps {
Actions = new [] { "sqs:SendMessage" },
Resources = new [] { targetQueue.QueueArn }
}));
var createScheduleTask1 = new EventBridgeSchedulerCreateScheduleTask(this, "createSchedule", new EventBridgeSchedulerCreateScheduleTaskProps {
ScheduleName = "TestSchedule",
ActionAfterCompletion = ActionAfterCompletion.NONE,
ClientToken = "testToken",
Description = "TestDescription",
StartDate = new Date(),
EndDate = new Date(new Date().GetTime() + 1000 * 60 * 60),
FlexibleTimeWindow = Duration.Minutes(5),
GroupName = scheduleGroup.Ref,
KmsKey = key,
Schedule = Schedule.Rate(Duration.Minutes(5)),
Timezone = "UTC",
Enabled = true,
Target = new EventBridgeSchedulerTarget(new EventBridgeSchedulerTargetProps {
Arn = targetQueue.QueueArn,
Role = schedulerRole,
RetryPolicy = new RetryPolicy {
MaximumRetryAttempts = 2,
MaximumEventAge = Duration.Minutes(5)
},
DeadLetterQueue = deadLetterQueue
})
});
go
import scheduler "github.com/aws/aws-cdk-go/awscdk"
import kms "github.com/aws/aws-cdk-go/awscdk"
var key Key
var scheduleGroup CfnScheduleGroup
var targetQueue Queue
var deadLetterQueue Queue
schedulerRole := iam.NewRole(this, jsii.String("SchedulerRole"), &RoleProps{
AssumedBy: iam.NewServicePrincipal(jsii.String("scheduler.amazonaws.com")),
})
// To send the message to the queue
// This policy changes depending on the type of target.
schedulerRole.AddToPrincipalPolicy(iam.NewPolicyStatement(&PolicyStatementProps{
Actions: []*string{
jsii.String("sqs:SendMessage"),
},
Resources: []*string{
targetQueue.queueArn,
},
}))
createScheduleTask1 := tasks.NewEventBridgeSchedulerCreateScheduleTask(this, jsii.String("createSchedule"), &EventBridgeSchedulerCreateScheduleTaskProps{
ScheduleName: jsii.String("TestSchedule"),
ActionAfterCompletion: tasks.ActionAfterCompletion_NONE,
ClientToken: jsii.String("testToken"),
Description: jsii.String("TestDescription"),
StartDate: NewDate(),
EndDate: NewDate(NewDate().getTime() + 1000 * 60 * 60),
FlexibleTimeWindow: awscdk.Duration_Minutes(jsii.Number(5)),
GroupName: scheduleGroup.ref,
KmsKey: key,
Schedule: tasks.Schedule_Rate(awscdk.Duration_*Minutes(jsii.Number(5))),
Timezone: jsii.String("UTC"),
Enabled: jsii.Boolean(true),
Target: tasks.NewEventBridgeSchedulerTarget(&EventBridgeSchedulerTargetProps{
Arn: targetQueue.queueArn,
Role: schedulerRole,
RetryPolicy: &RetryPolicy{
MaximumRetryAttempts: jsii.Number(2),
MaximumEventAge: awscdk.Duration_*Minutes(jsii.Number(5)),
},
DeadLetterQueue: *DeadLetterQueue,
}),
})
Glue
Step Functions supports AWS Glue through the service integration pattern.
StartJobRun
You can call the StartJobRun API from a Task state.
ts
new tasks.GlueStartJobRun(this, 'Task', {
glueJobName: 'my-glue-job',
arguments: sfn.TaskInput.fromObject({
key: 'value',
}),
taskTimeout: sfn.Timeout.duration(Duration.minutes(30)),
notifyDelayAfter: Duration.minutes(5),
});
python
tasks.GlueStartJobRun(self, "Task",
glue_job_name="my-glue-job",
arguments=sfn.TaskInput.from_object({
"key": "value"
}),
task_timeout=sfn.Timeout.duration(Duration.minutes(30)),
notify_delay_after=Duration.minutes(5)
)
java
GlueStartJobRun.Builder.create(this, "Task")
.glueJobName("my-glue-job")
.arguments(TaskInput.fromObject(Map.of(
"key", "value")))
.taskTimeout(Timeout.duration(Duration.minutes(30)))
.notifyDelayAfter(Duration.minutes(5))
.build();
csharp
new GlueStartJobRun(this, "Task", new GlueStartJobRunProps {
GlueJobName = "my-glue-job",
Arguments = TaskInput.FromObject(new Dictionary<string, object> {
{ "key", "value" }
}),
TaskTimeout = Timeout.Duration(Duration.Minutes(30)),
NotifyDelayAfter = Duration.Minutes(5)
});
go
tasks.NewGlueStartJobRun(this, jsii.String("Task"), &GlueStartJobRunProps{
GlueJobName: jsii.String("my-glue-job"),
Arguments: sfn.TaskInput_FromObject(map[string]interface{}{
"key": jsii.String("value"),
}),
TaskTimeout: sfn.Timeout_Duration(awscdk.Duration_Minutes(jsii.Number(30))),
NotifyDelayAfter: awscdk.Duration_*Minutes(jsii.Number(5)),
})
You can configure workers by setting the workerTypeV2 and numberOfWorkers properties. workerType is deprecated and no longer recommended. Use workerTypeV2 which is a ENUM-like class for more powerful worker configuration around using pre-defined values or dynamic values.
ts
new tasks.GlueStartJobRun(this, 'Task', {
glueJobName: 'my-glue-job',
workerConfiguration: {
workerTypeV2: tasks.WorkerTypeV2.G_1X, // Worker type
numberOfWorkers: 2, // Number of Workers
},
});
python
tasks.GlueStartJobRun(self, "Task",
glue_job_name="my-glue-job",
worker_configuration=tasks.WorkerConfigurationProperty(
worker_type_v2=tasks.WorkerTypeV2.G_1X, # Worker type
number_of_workers=2
)
)
java
GlueStartJobRun.Builder.create(this, "Task")
.glueJobName("my-glue-job")
.workerConfiguration(WorkerConfigurationProperty.builder()
.workerTypeV2(WorkerTypeV2.G_1X) // Worker type
.numberOfWorkers(2)
.build())
.build();
csharp
new GlueStartJobRun(this, "Task", new GlueStartJobRunProps {
GlueJobName = "my-glue-job",
WorkerConfiguration = new WorkerConfigurationProperty {
WorkerTypeV2 = WorkerTypeV2.G_1X, // Worker type
NumberOfWorkers = 2
}
});
go
tasks.NewGlueStartJobRun(this, jsii.String("Task"), &GlueStartJobRunProps{
GlueJobName: jsii.String("my-glue-job"),
WorkerConfiguration: &WorkerConfigurationProperty{
WorkerTypeV2: tasks.WorkerTypeV2_G_1X(),
// Worker type
NumberOfWorkers: jsii.Number(2),
},
})
To configure the worker type or number of workers dynamically from StateMachine's input, you can configure it using JSON Path values using workerTypeV2 like this:
ts
new tasks.GlueStartJobRun(this, 'Glue Job Task', {
glueJobName: 'my-glue-job',
workerConfiguration: {
workerTypeV2: tasks.WorkerTypeV2.of(sfn.JsonPath.stringAt('$.glue_jobs_configs.executor_type')),
numberOfWorkers: sfn.JsonPath.numberAt('$.glue_jobs_configs.max_number_workers'),
},
});
python
tasks.GlueStartJobRun(self, "Glue Job Task",
glue_job_name="my-glue-job",
worker_configuration=tasks.WorkerConfigurationProperty(
worker_type_v2=tasks.WorkerTypeV2.of(sfn.JsonPath.string_at("$.glue_jobs_configs.executor_type")),
number_of_workers=sfn.JsonPath.number_at("$.glue_jobs_configs.max_number_workers")
)
)
java
GlueStartJobRun.Builder.create(this, "Glue Job Task")
.glueJobName("my-glue-job")
.workerConfiguration(WorkerConfigurationProperty.builder()
.workerTypeV2(WorkerTypeV2.of(JsonPath.stringAt("$.glue_jobs_configs.executor_type")))
.numberOfWorkers(JsonPath.numberAt("$.glue_jobs_configs.max_number_workers"))
.build())
.build();
csharp
new GlueStartJobRun(this, "Glue Job Task", new GlueStartJobRunProps {
GlueJobName = "my-glue-job",
WorkerConfiguration = new WorkerConfigurationProperty {
WorkerTypeV2 = WorkerTypeV2.Of(JsonPath.StringAt("$.glue_jobs_configs.executor_type")),
NumberOfWorkers = JsonPath.NumberAt("$.glue_jobs_configs.max_number_workers")
}
});
go
tasks.NewGlueStartJobRun(this, jsii.String("Glue Job Task"), &GlueStartJobRunProps{
GlueJobName: jsii.String("my-glue-job"),
WorkerConfiguration: &WorkerConfigurationProperty{
WorkerTypeV2: tasks.WorkerTypeV2_Of(sfn.JsonPath_StringAt(jsii.String("$.glue_jobs_configs.executor_type"))),
NumberOfWorkers: sfn.JsonPath_NumberAt(jsii.String("$.glue_jobs_configs.max_number_workers")),
},
})
You can choose the execution class by setting the executionClass property.
ts
new tasks.GlueStartJobRun(this, 'Task', {
glueJobName: 'my-glue-job',
executionClass: tasks.ExecutionClass.FLEX,
});
python
tasks.GlueStartJobRun(self, "Task",
glue_job_name="my-glue-job",
execution_class=tasks.ExecutionClass.FLEX
)
java
GlueStartJobRun.Builder.create(this, "Task")
.glueJobName("my-glue-job")
.executionClass(ExecutionClass.FLEX)
.build();
csharp
new GlueStartJobRun(this, "Task", new GlueStartJobRunProps {
GlueJobName = "my-glue-job",
ExecutionClass = ExecutionClass.FLEX
});
go
tasks.NewGlueStartJobRun(this, jsii.String("Task"), &GlueStartJobRunProps{
GlueJobName: jsii.String("my-glue-job"),
ExecutionClass: tasks.ExecutionClass_FLEX,
})
StartCrawlerRun
You can call the StartCrawler API from a Task state through AWS SDK service integrations.
ts
import * as glue from 'aws-cdk-lib/aws-glue';
declare const myCrawler: glue.CfnCrawler;
// You can get the crawler name from `crawler.ref`
new tasks.GlueStartCrawlerRun(this, 'Task1', {
crawlerName: myCrawler.ref,
});
// Of course, you can also specify the crawler name directly.
new tasks.GlueStartCrawlerRun(this, 'Task2', {
crawlerName: 'my-crawler-job',
});
python
import aws_cdk.aws_glue as glue
# my_crawler: glue.CfnCrawler
# You can get the crawler name from `crawler.ref`
tasks.GlueStartCrawlerRun(self, "Task1",
crawler_name=my_crawler.ref
)
# Of course, you can also specify the crawler name directly.
tasks.GlueStartCrawlerRun(self, "Task2",
crawler_name="my-crawler-job"
)
java
import software.amazon.awscdk.services.glue.*;
CfnCrawler myCrawler;
// You can get the crawler name from `crawler.ref`
// You can get the crawler name from `crawler.ref`
GlueStartCrawlerRun.Builder.create(this, "Task1")
.crawlerName(myCrawler.getRef())
.build();
// Of course, you can also specify the crawler name directly.
// Of course, you can also specify the crawler name directly.
GlueStartCrawlerRun.Builder.create(this, "Task2")
.crawlerName("my-crawler-job")
.build();
csharp
using Amazon.CDK.AWS.Glue;
CfnCrawler myCrawler;
// You can get the crawler name from `crawler.ref`
// You can get the crawler name from `crawler.ref`
new GlueStartCrawlerRun(this, "Task1", new GlueStartCrawlerRunProps {
CrawlerName = myCrawler.Ref
});
// Of course, you can also specify the crawler name directly.
// Of course, you can also specify the crawler name directly.
new GlueStartCrawlerRun(this, "Task2", new GlueStartCrawlerRunProps {
CrawlerName = "my-crawler-job"
});
go
import glue "github.com/aws/aws-cdk-go/awscdk"
var myCrawler CfnCrawler
// You can get the crawler name from `crawler.ref`
// You can get the crawler name from `crawler.ref`
tasks.NewGlueStartCrawlerRun(this, jsii.String("Task1"), &GlueStartCrawlerRunProps{
CrawlerName: myCrawler.ref,
})
// Of course, you can also specify the crawler name directly.
// Of course, you can also specify the crawler name directly.
tasks.NewGlueStartCrawlerRun(this, jsii.String("Task2"), &GlueStartCrawlerRunProps{
CrawlerName: jsii.String("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.
ts
new tasks.GlueDataBrewStartJobRun(this, 'Task', {
name: 'databrew-job',
});
python
tasks.GlueDataBrewStartJobRun(self, "Task",
name="databrew-job"
)
java
GlueDataBrewStartJobRun.Builder.create(this, "Task")
.name("databrew-job")
.build();
csharp
new GlueDataBrewStartJobRun(this, "Task", new GlueDataBrewStartJobRunProps {
Name = "databrew-job"
});
go
tasks.NewGlueDataBrewStartJobRun(this, jsii.String("Task"), &GlueDataBrewStartJobRunProps{
Name: jsii.String("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.
ts
import * as events from 'aws-cdk-lib/aws-events';
const connection = new events.Connection(this, 'Connection', {
authorization: events.Authorization.basic('username', SecretValue.unsafePlainText('password')),
});
new tasks.HttpInvoke(this, 'Invoke HTTP API', {
apiRoot: 'https://api.example.com',
apiEndpoint: sfn.TaskInput.fromText(sfn.JsonPath.format('resource/{}/details', sfn.JsonPath.stringAt('$.resourceId'))),
body: sfn.TaskInput.fromObject({ foo: 'bar' }),
connection,
headers: sfn.TaskInput.fromObject({ 'Content-Type': 'application/json' }),
method: sfn.TaskInput.fromText('POST'),
queryStringParameters: sfn.TaskInput.fromObject({ id: '123' }),
urlEncodingFormat: tasks.URLEncodingFormat.BRACKETS,
});
python
import aws_cdk.aws_events as events
connection = events.Connection(self, "Connection",
authorization=events.Authorization.basic("username", SecretValue.unsafe_plain_text("password"))
)
tasks.HttpInvoke(self, "Invoke HTTP API",
api_root="https://api.example.com",
api_endpoint=sfn.TaskInput.from_text(sfn.JsonPath.format("resource/{}/details", sfn.JsonPath.string_at("$.resourceId"))),
body=sfn.TaskInput.from_object({"foo": "bar"}),
connection=connection,
headers=sfn.TaskInput.from_object({"Content-Type": "application/json"}),
method=sfn.TaskInput.from_text("POST"),
query_string_parameters=sfn.TaskInput.from_object({"id": "123"}),
url_encoding_format=tasks.URLEncodingFormat.BRACKETS
)
java
import software.amazon.awscdk.services.events.*;
Connection connection = Connection.Builder.create(this, "Connection")
.authorization(Authorization.basic("username", SecretValue.unsafePlainText("password")))
.build();
HttpInvoke.Builder.create(this, "Invoke HTTP API")
.apiRoot("https://api.example.com")
.apiEndpoint(TaskInput.fromText(JsonPath.format("resource/{}/details", JsonPath.stringAt("$.resourceId"))))
.body(TaskInput.fromObject(Map.of("foo", "bar")))
.connection(connection)
.headers(TaskInput.fromObject(Map.of("Content-Type", "application/json")))
.method(TaskInput.fromText("POST"))
.queryStringParameters(TaskInput.fromObject(Map.of("id", "123")))
.urlEncodingFormat(URLEncodingFormat.BRACKETS)
.build();
csharp
using Amazon.CDK.AWS.Events;
var connection = new Connection(this, "Connection", new ConnectionProps {
Authorization = Authorization.Basic("username", SecretValue.UnsafePlainText("password"))
});
new HttpInvoke(this, "Invoke HTTP API", new HttpInvokeProps {
ApiRoot = "https://api.example.com",
ApiEndpoint = TaskInput.FromText(JsonPath.Format("resource/{}/details", JsonPath.StringAt("$.resourceId"))),
Body = TaskInput.FromObject(new Dictionary<string, object> { { "foo", "bar" } }),
Connection = connection,
Headers = TaskInput.FromObject(new Dictionary<string, object> { { "Content-Type", "application/json" } }),
Method = TaskInput.FromText("POST"),
QueryStringParameters = TaskInput.FromObject(new Dictionary<string, object> { { "id", "123" } }),
UrlEncodingFormat = URLEncodingFormat.BRACKETS
});
go
import "github.com/aws/aws-cdk-go/awscdk"
connection := events.NewConnection(this, jsii.String("Connection"), &ConnectionProps{
Authorization: events.Authorization_Basic(jsii.String("username"), awscdk.SecretValue_UnsafePlainText(jsii.String("password"))),
})
tasks.NewHttpInvoke(this, jsii.String("Invoke HTTP API"), &HttpInvokeProps{
ApiRoot: jsii.String("https://api.example.com"),
ApiEndpoint: sfn.TaskInput_FromText(sfn.JsonPath_Format(jsii.String("resource/{}/details"), sfn.JsonPath_StringAt(jsii.String("$.resourceId")))),
Body: sfn.TaskInput_FromObject(map[string]interface{}{
"foo": jsii.String("bar"),
}),
Connection: Connection,
Headers: sfn.TaskInput_*FromObject(map[string]interface{}{
"Content-Type": jsii.String("application/json"),
}),
Method: sfn.TaskInput_*FromText(jsii.String("POST")),
QueryStringParameters: sfn.TaskInput_*FromObject(map[string]interface{}{
"id": jsii.String("123"),
}),
UrlEncodingFormat: tasks.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.
ts
declare const fn: lambda.Function;
new tasks.LambdaInvoke(this, 'Invoke with state input', {
lambdaFunction: fn,
});
python
# fn: lambda.Function
tasks.LambdaInvoke(self, "Invoke with state input",
lambda_function=fn
)
java
Function fn;
LambdaInvoke.Builder.create(this, "Invoke with state input")
.lambdaFunction(fn)
.build();
csharp
Function fn;
new LambdaInvoke(this, "Invoke with state input", new LambdaInvokeProps {
LambdaFunction = fn
});
go
var fn Function
tasks.NewLambdaInvoke(this, jsii.String("Invoke with state input"), &LambdaInvokeProps{
LambdaFunction: 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.
ts
declare const fn: lambda.Function;
new tasks.LambdaInvoke(this, 'Invoke with empty object as payload', {
lambdaFunction: fn,
payload: sfn.TaskInput.fromObject({}),
});
// use the output of fn as input
new tasks.LambdaInvoke(this, 'Invoke with payload field in the state input', {
lambdaFunction: fn,
payload: sfn.TaskInput.fromJsonPathAt('$.Payload'),
});
python
# fn: lambda.Function
tasks.LambdaInvoke(self, "Invoke with empty object as payload",
lambda_function=fn,
payload=sfn.TaskInput.from_object({})
)
# use the output of fn as input
tasks.LambdaInvoke(self, "Invoke with payload field in the state input",
lambda_function=fn,
payload=sfn.TaskInput.from_json_path_at("$.Payload")
)
java
Function fn;
LambdaInvoke.Builder.create(this, "Invoke with empty object as payload")
.lambdaFunction(fn)
.payload(TaskInput.fromObject(Map.of()))
.build();
// use the output of fn as input
// use the output of fn as input
LambdaInvoke.Builder.create(this, "Invoke with payload field in the state input")
.lambdaFunction(fn)
.payload(TaskInput.fromJsonPathAt("$.Payload"))
.build();
csharp
Function fn;
new LambdaInvoke(this, "Invoke with empty object as payload", new LambdaInvokeProps {
LambdaFunction = fn,
Payload = TaskInput.FromObject(new Dictionary<string, object> { })
});
// use the output of fn as input
// use the output of fn as input
new LambdaInvoke(this, "Invoke with payload field in the state input", new LambdaInvokeProps {
LambdaFunction = fn,
Payload = TaskInput.FromJsonPathAt("$.Payload")
});
go
var fn Function
tasks.NewLambdaInvoke(this, jsii.String("Invoke with empty object as payload"), &LambdaInvokeProps{
LambdaFunction: fn,
Payload: sfn.TaskInput_FromObject(map[string]interface{}{
}),
})
// use the output of fn as input
// use the output of fn as input
tasks.NewLambdaInvoke(this, jsii.String("Invoke with payload field in the state input"), &LambdaInvokeProps{
LambdaFunction: fn,
Payload: sfn.TaskInput_FromJsonPathAt(jsii.String("$.Payload")),
})
The following snippet invokes a Lambda and sets the task output to only include the Lambda function response.
ts
declare const fn: lambda.Function;
new tasks.LambdaInvoke(this, 'Invoke and set function response as task output', {
lambdaFunction: fn,
outputPath: '$.Payload',
});
python
# fn: lambda.Function
tasks.LambdaInvoke(self, "Invoke and set function response as task output",
lambda_function=fn,
output_path="$.Payload"
)
java
Function fn;
LambdaInvoke.Builder.create(this, "Invoke and set function response as task output")
.lambdaFunction(fn)
.outputPath("$.Payload")
.build();
csharp
Function fn;
new LambdaInvoke(this, "Invoke and set function response as task output", new LambdaInvokeProps {
LambdaFunction = fn,
OutputPath = "$.Payload"
});
go
var fn Function
tasks.NewLambdaInvoke(this, jsii.String("Invoke and set function response as task output"), &LambdaInvokeProps{
LambdaFunction: fn,
OutputPath: jsii.String("$.Payload"),
})
If you want to combine the input and the Lambda function response you can use the payloadResponseOnly property and specify the resultPath. This will put the Lambda function ARN directly in the "Resource" string, but it conflicts with the integrationPattern, invocationType, clientContext, and qualifier properties.
ts
declare const fn: lambda.Function;
new tasks.LambdaInvoke(this, 'Invoke and combine function response with task input', {
lambdaFunction: fn,
payloadResponseOnly: true,
resultPath: '$.fn',
});
python
# fn: lambda.Function
tasks.LambdaInvoke(self, "Invoke and combine function response with task input",
lambda_function=fn,
payload_response_only=True,
result_path="$.fn"
)
java
Function fn;
LambdaInvoke.Builder.create(this, "Invoke and combine function response with task input")
.lambdaFunction(fn)
.payloadResponseOnly(true)
.resultPath("$.fn")
.build();
csharp
Function fn;
new LambdaInvoke(this, "Invoke and combine function response with task input", new LambdaInvokeProps {
LambdaFunction = fn,
PayloadResponseOnly = true,
ResultPath = "$.fn"
});
go
var fn Function
tasks.NewLambdaInvoke(this, jsii.String("Invoke and combine function response with task input"), &LambdaInvokeProps{
LambdaFunction: fn,
PayloadResponseOnly: jsii.Boolean(true),
ResultPath: jsii.String("$.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.
ts
declare const fn: lambda.Function;
new tasks.LambdaInvoke(this, 'Invoke with callback', {
lambdaFunction: fn,
integrationPattern: sfn.IntegrationPattern.WAIT_FOR_TASK_TOKEN,
payload: sfn.TaskInput.fromObject({
token: sfn.JsonPath.taskToken,
input: sfn.JsonPath.stringAt('$.someField'),
}),
});
python
# fn: lambda.Function
tasks.LambdaInvoke(self, "Invoke with callback",
lambda_function=fn,
integration_pattern=sfn.IntegrationPattern.WAIT_FOR_TASK_TOKEN,
payload=sfn.TaskInput.from_object({
"token": sfn.JsonPath.task_token,
"input": sfn.JsonPath.string_at("$.someField")
})
)
java
Function fn;
LambdaInvoke.Builder.create(this, "Invoke with callback")
.lambdaFunction(fn)
.integrationPattern(IntegrationPattern.WAIT_FOR_TASK_TOKEN)
.payload(TaskInput.fromObject(Map.of(
"token", JsonPath.getTaskToken(),
"input", JsonPath.stringAt("$.someField"))))
.build();
csharp
Function fn;
new LambdaInvoke(this, "Invoke with callback", new LambdaInvokeProps {
LambdaFunction = fn,
IntegrationPattern = IntegrationPattern.WAIT_FOR_TASK_TOKEN,
Payload = TaskInput.FromObject(new Dictionary<string, object> {
{ "token", JsonPath.TaskToken },
{ "input", JsonPath.StringAt("$.someField") }
})
});
go
var fn Function
tasks.NewLambdaInvoke(this, jsii.String("Invoke with callback"), &LambdaInvokeProps{
LambdaFunction: fn,
IntegrationPattern: sfn.IntegrationPattern_WAIT_FOR_TASK_TOKEN,
Payload: sfn.TaskInput_FromObject(map[string]interface{}{
"token": sfn.JsonPath_taskToken(),
"input": sfn.JsonPath_stringAt(jsii.String("$.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 retryOnServiceExceptions 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 integrationPattern.
Make sure you update the required fields - Role & Settings and refer CreateJobRequest for all other optional parameters.
ts
new tasks.MediaConvertCreateJob(this, 'CreateJob', {
createJobRequest: {
"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"
}
]
}
},
integrationPattern: sfn.IntegrationPattern.RUN_JOB,
});
python
tasks.MediaConvertCreateJob(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=sfn.IntegrationPattern.RUN_JOB
)
java
MediaConvertCreateJob.Builder.create(this, "CreateJob")
.createJobRequest(Map.of(
"Role", "arn:aws:iam::123456789012:role/MediaConvertRole",
"Settings", Map.of(
"OutputGroups", List.of(Map.of(
"Outputs", List.of(Map.of(
"ContainerSettings", Map.of(
"Container", "MP4"),
"VideoDescription", Map.of(
"CodecSettings", Map.of(
"Codec", "H_264",
"H264Settings", Map.of(
"MaxBitrate", 1000,
"RateControlMode", "QVBR",
"SceneChangeDetect", "TRANSITION_DETECTION"))),
"AudioDescriptions", List.of(Map.of(
"CodecSettings", Map.of(
"Codec", "AAC",
"AacSettings", Map.of(
"Bitrate", 96000,
"CodingMode", "CODING_MODE_2_0",
"SampleRate", 48000)))))),
"OutputGroupSettings", Map.of(
"Type", "FILE_GROUP_SETTINGS",
"FileGroupSettings", Map.of(
"Destination", "s3://EXAMPLE-DESTINATION-BUCKET/")))),
"Inputs", List.of(Map.of(
"AudioSelectors", Map.of(
"Audio Selector 1", Map.of(
"DefaultSelection", "DEFAULT")),
"FileInput", "s3://EXAMPLE-SOURCE-BUCKET/EXAMPLE-SOURCE_FILE")))))
.integrationPattern(IntegrationPattern.RUN_JOB)
.build();
csharp
new MediaConvertCreateJob(this, "CreateJob", new MediaConvertCreateJobProps {
CreateJobRequest = new Dictionary<string, object> {
{ "Role", "arn:aws:iam::123456789012:role/MediaConvertRole" },
{ "Settings", new Dictionary<string, IDictionary<string, object>[]> {
{ "OutputGroups", new [] { new Struct {
Outputs = new [] { new Struct {
ContainerSettings = new Struct {
Container = "MP4"
},
VideoDescription = new Struct {
CodecSettings = new Struct {
Codec = "H_264",
H264Settings = new Struct {
MaxBitrate = 1000,
RateControlMode = "QVBR",
SceneChangeDetect = "TRANSITION_DETECTION"
}
}
},
AudioDescriptions = new [] { new Struct {
CodecSettings = new Struct {
Codec = "AAC",
AacSettings = new Struct {
Bitrate = 96000,
CodingMode = "CODING_MODE_2_0",
SampleRate = 48000
}
}
} }
} },
OutputGroupSettings = new Struct {
Type = "FILE_GROUP_SETTINGS",
FileGroupSettings = new Struct {
Destination = "s3://EXAMPLE-DESTINATION-BUCKET/"
}
}
} } },
{ "Inputs", new [] { new Struct {
AudioSelectors = new Struct {
Audio Selector 1 = new Struct {
DefaultSelection = "DEFAULT"
}
},
FileInput = "s3://EXAMPLE-SOURCE-BUCKET/EXAMPLE-SOURCE_FILE"
} } }
} }
},
IntegrationPattern = IntegrationPattern.RUN_JOB
});
go
tasks.NewMediaConvertCreateJob(this, jsii.String("CreateJob"), &MediaConvertCreateJobProps{
CreateJobRequest: map[string]interface{}{
"Role": jsii.String("arn:aws:iam::123456789012:role/MediaConvertRole"),
"Settings": map[string][]map[string]interface{}{
"OutputGroups": []map[string]interface{}{
map[string]interface{}{
"Outputs": []map[string]interface{}{
map[string]interface{}{
"ContainerSettings": map[string]*string{
"Container": jsii.String("MP4"),
},
"VideoDescription": map[string]map[string]interface{}{
"CodecSettings": map[string]interface{}{
"Codec": jsii.String("H_264"),
"H264Settings": map[string]interface{}{
"MaxBitrate": jsii.Number(1000),
"RateControlMode": jsii.String("QVBR"),
"SceneChangeDetect": jsii.String("TRANSITION_DETECTION"),
},
},
},
"AudioDescriptions": []map[string]map[string]interface{}{
map[string]map[string]interface{}{
"CodecSettings": map[string]interface{}{
"Codec": jsii.String("AAC"),
"AacSettings": map[string]interface{}{
"Bitrate": jsii.Number(96000),
"CodingMode": jsii.String("CODING_MODE_2_0"),
"SampleRate": jsii.Number(48000),
},
},
},
},
},
},
"OutputGroupSettings": map[string]interface{}{
"Type": jsii.String("FILE_GROUP_SETTINGS"),
"FileGroupSettings": map[string]*string{
"Destination": jsii.String("s3://EXAMPLE-DESTINATION-BUCKET/"),
},
},
},
},
"Inputs": []map[string]interface{}{
map[string]interface{}{
"AudioSelectors": map[string]map[string]*string{
"Audio Selector 1": map[string]*string{
"DefaultSelection": jsii.String("DEFAULT"),
},
},
"FileInput": jsii.String("s3://EXAMPLE-SOURCE-BUCKET/EXAMPLE-SOURCE_FILE"),
},
},
},
},
IntegrationPattern: sfn.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 enableNetworkIsolation 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.
ts
new tasks.SageMakerCreateTrainingJob(this, 'TrainSagemaker', {
trainingJobName: sfn.JsonPath.stringAt('$.JobName'),
algorithmSpecification: {
algorithmName: 'BlazingText',
trainingInputMode: tasks.InputMode.FILE,
},
inputDataConfig: [{
channelName: 'train',
dataSource: {
s3DataSource: {
s3DataType: tasks.S3DataType.S3_PREFIX,
s3Location: tasks.S3Location.fromJsonExpression('$.S3Bucket'),
},
},
}],
outputDataConfig: {
s3OutputLocation: tasks.S3Location.fromBucket(s3.Bucket.fromBucketName(this, 'Bucket', 'amzn-s3-demo-bucket'), 'myoutputpath'),
},
resourceConfig: {
instanceCount: 1,
instanceType: new ec2.InstanceType(sfn.JsonPath.stringAt('$.InstanceType')),
volumeSize: Size.gibibytes(50),
}, // optional: default is 1 instance of EC2 `M4.XLarge` with `10GB` volume
stoppingCondition: {
maxRuntime: Duration.hours(2),
}, // optional: default is 1 hour
});
python
tasks.SageMakerCreateTrainingJob(self, "TrainSagemaker",
training_job_name=sfn.JsonPath.string_at("$.JobName"),
algorithm_specification=tasks.AlgorithmSpecification(
algorithm_name="BlazingText",
training_input_mode=tasks.InputMode.FILE
),
input_data_config=[tasks.Channel(
channel_name="train",
data_source=tasks.DataSource(
s3_data_source=tasks.S3DataSource(
s3_data_type=tasks.S3DataType.S3_PREFIX,
s3_location=tasks.S3Location.from_json_expression("$.S3Bucket")
)
)
)],
output_data_config=tasks.OutputDataConfig(
s3_output_location=tasks.S3Location.from_bucket(s3.Bucket.from_bucket_name(self, "Bucket", "amzn-s3-demo-bucket"), "myoutputpath")
),
resource_config=tasks.ResourceConfig(
instance_count=1,
instance_type=ec2.InstanceType(sfn.JsonPath.string_at("$.InstanceType")),
volume_size=Size.gibibytes(50)
), # optional: default is 1 instance of EC2 `M4.XLarge` with `10GB` volume
stopping_condition=tasks.StoppingCondition(
max_runtime=Duration.hours(2)
)
)
java
SageMakerCreateTrainingJob.Builder.create(this, "TrainSagemaker")
.trainingJobName(JsonPath.stringAt("$.JobName"))
.algorithmSpecification(AlgorithmSpecification.builder()
.algorithmName("BlazingText")
.trainingInputMode(InputMode.FILE)
.build())
.inputDataConfig(List.of(Channel.builder()
.channelName("train")
.dataSource(DataSource.builder()
.s3DataSource(S3DataSource.builder()
.s3DataType(S3DataType.S3_PREFIX)
.s3Location(S3Location.fromJsonExpression("$.S3Bucket"))
.build())
.build())
.build()))
.outputDataConfig(OutputDataConfig.builder()
.s3OutputLocation(S3Location.fromBucket(Bucket.fromBucketName(this, "Bucket", "amzn-s3-demo-bucket"), "myoutputpath"))
.build())
.resourceConfig(ResourceConfig.builder()
.instanceCount(1)
.instanceType(new InstanceType(JsonPath.stringAt("$.InstanceType")))
.volumeSize(Size.gibibytes(50))
.build()) // optional: default is 1 instance of EC2 `M4.XLarge` with `10GB` volume
.stoppingCondition(StoppingCondition.builder()
.maxRuntime(Duration.hours(2))
.build())
.build();
csharp
new SageMakerCreateTrainingJob(this, "TrainSagemaker", new SageMakerCreateTrainingJobProps {
TrainingJobName = JsonPath.StringAt("$.JobName"),
AlgorithmSpecification = new AlgorithmSpecification {
AlgorithmName = "BlazingText",
TrainingInputMode = InputMode.FILE
},
InputDataConfig = new [] { new Channel {
ChannelName = "train",
DataSource = new DataSource {
S3DataSource = new S3DataSource {
S3DataType = S3DataType.S3_PREFIX,
S3Location = S3Location.FromJsonExpression("$.S3Bucket")
}
}
} },
OutputDataConfig = new OutputDataConfig {
S3OutputLocation = S3Location.FromBucket(Bucket.FromBucketName(this, "Bucket", "amzn-s3-demo-bucket"), "myoutputpath")
},
ResourceConfig = new ResourceConfig {
InstanceCount = 1,
InstanceType = new InstanceType(JsonPath.StringAt("$.InstanceType")),
VolumeSize = Size.Gibibytes(50)
}, // optional: default is 1 instance of EC2 `M4.XLarge` with `10GB` volume
StoppingCondition = new StoppingCondition {
MaxRuntime = Duration.Hours(2)
}
});
go
tasks.NewSageMakerCreateTrainingJob(this, jsii.String("TrainSagemaker"), &SageMakerCreateTrainingJobProps{
TrainingJobName: sfn.JsonPath_StringAt(jsii.String("$.JobName")),
AlgorithmSpecification: &AlgorithmSpecification{
AlgorithmName: jsii.String("BlazingText"),
TrainingInputMode: tasks.InputMode_FILE,
},
InputDataConfig: []Channel{
&Channel{
ChannelName: jsii.String("train"),
DataSource: &DataSource{
S3DataSource: &S3DataSource{
S3DataType: tasks.S3DataType_S3_PREFIX,
S3Location: tasks.S3Location_FromJsonExpression(jsii.String("$.S3Bucket")),
},
},
},
},
OutputDataConfig: &OutputDataConfig{
S3OutputLocation: tasks.S3Location_FromBucket(s3.Bucket_FromBucketName(this, jsii.String("Bucket"), jsii.String("amzn-s3-demo-bucket")), jsii.String("myoutputpath")),
},
ResourceConfig: &ResourceConfig{
InstanceCount: jsii.Number(1),
InstanceType: ec2.NewInstanceType(sfn.JsonPath_*StringAt(jsii.String("$.InstanceType"))),
VolumeSize: awscdk.Size_Gibibytes(jsii.Number(50)),
},
// optional: default is 1 instance of EC2 `M4.XLarge` with `10GB` volume
StoppingCondition: &StoppingCondition{
MaxRuntime: awscdk.Duration_Hours(jsii.Number(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.FILEif an algorithm supports it. - To stream data directly from Amazon S3 to the container, choose
InputMode.PIPEif 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_FILEif an algorithm supports it.
Create Transform Job
You can call the CreateTransformJob API from a Task state.
ts
new tasks.SageMakerCreateTransformJob(this, 'Batch Inference', {
transformJobName: 'MyTransformJob',
modelName: 'MyModelName',
modelClientOptions: {
invocationsMaxRetries: 3, // default is 0
invocationsTimeout: Duration.minutes(5), // default is 60 seconds
},
transformInput: {
transformDataSource: {
s3DataSource: {
s3Uri: 's3://inputbucket/train',
s3DataType: tasks.S3DataType.S3_PREFIX,
}
}
},
transformOutput: {
s3OutputPath: 's3://outputbucket/TransformJobOutputPath',
},
transformResources: {
instanceCount: 1,
instanceType: ec2.InstanceType.of(ec2.InstanceClass.M4, ec2.InstanceSize.XLARGE),
}
});
python
tasks.SageMakerCreateTransformJob(self, "Batch Inference",
transform_job_name="MyTransformJob",
model_name="MyModelName",
model_client_options=tasks.ModelClientOptions(
invocations_max_retries=3, # default is 0
invocations_timeout=Duration.minutes(5)
),
transform_input=tasks.TransformInput(
transform_data_source=tasks.TransformDataSource(
s3_data_source=tasks.TransformS3DataSource(
s3_uri="s3://inputbucket/train",
s3_data_type=tasks.S3DataType.S3_PREFIX
)
)
),
transform_output=tasks.TransformOutput(
s3_output_path="s3://outputbucket/TransformJobOutputPath"
),
transform_resources=tasks.TransformResources(
instance_count=1,
instance_type=ec2.InstanceType.of(ec2.InstanceClass.M4, ec2.InstanceSize.XLARGE)
)
)
java
SageMakerCreateTransformJob.Builder.create(this, "Batch Inference")
.transformJobName("MyTransformJob")
.modelName("MyModelName")
.modelClientOptions(ModelClientOptions.builder()
.invocationsMaxRetries(3) // default is 0
.invocationsTimeout(Duration.minutes(5))
.build())
.transformInput(TransformInput.builder()
.transformDataSource(TransformDataSource.builder()
.s3DataSource(TransformS3DataSource.builder()
.s3Uri("s3://inputbucket/train")
.s3DataType(S3DataType.S3_PREFIX)
.build())
.build())
.build())
.transformOutput(TransformOutput.builder()
.s3OutputPath("s3://outputbucket/TransformJobOutputPath")
.build())
.transformResources(TransformResources.builder()
.instanceCount(1)
.instanceType(InstanceType.of(InstanceClass.M4, InstanceSize.XLARGE))
.build())
.build();
csharp
new SageMakerCreateTransformJob(this, "Batch Inference", new SageMakerCreateTransformJobProps {
TransformJobName = "MyTransformJob",
ModelName = "MyModelName",
ModelClientOptions = new ModelClientOptions {
InvocationsMaxRetries = 3, // default is 0
InvocationsTimeout = Duration.Minutes(5)
},
TransformInput = new TransformInput {
TransformDataSource = new TransformDataSource {
S3DataSource = new TransformS3DataSource {
S3Uri = "s3://inputbucket/train",
S3DataType = S3DataType.S3_PREFIX
}
}
},
TransformOutput = new TransformOutput {
S3OutputPath = "s3://outputbucket/TransformJobOutputPath"
},
TransformResources = new TransformResources {
InstanceCount = 1,
InstanceType = InstanceType.Of(InstanceClass.M4, InstanceSize.XLARGE)
}
});
go
tasks.NewSageMakerCreateTransformJob(this, jsii.String("Batch Inference"), &SageMakerCreateTransformJobProps{
TransformJobName: jsii.String("MyTransformJob"),
ModelName: jsii.String("MyModelName"),
ModelClientOptions: &ModelClientOptions{
InvocationsMaxRetries: jsii.Number(3),
// default is 0
InvocationsTimeout: awscdk.Duration_Minutes(jsii.Number(5)),
},
TransformInput: &TransformInput{
TransformDataSource: &TransformDataSource{
S3DataSource: &TransformS3DataSource{
S3Uri: jsii.String("s3://inputbucket/train"),
S3DataType: tasks.S3DataType_S3_PREFIX,
},
},
},
TransformOutput: &TransformOutput{
S3OutputPath: jsii.String("s3://outputbucket/TransformJobOutputPath"),
},
TransformResources: &TransformResources{
InstanceCount: jsii.Number(1),
InstanceType: ec2.InstanceType_Of(ec2.InstanceClass_M4, ec2.InstanceSize_XLARGE),
},
})
Create Endpoint
You can call the CreateEndpoint API from a Task state.
ts
new tasks.SageMakerCreateEndpoint(this, 'SagemakerEndpoint', {
endpointName: sfn.JsonPath.stringAt('$.EndpointName'),
endpointConfigName: sfn.JsonPath.stringAt('$.EndpointConfigName'),
});
python
tasks.SageMakerCreateEndpoint(self, "SagemakerEndpoint",
endpoint_name=sfn.JsonPath.string_at("$.EndpointName"),
endpoint_config_name=sfn.JsonPath.string_at("$.EndpointConfigName")
)
java
SageMakerCreateEndpoint.Builder.create(this, "SagemakerEndpoint")
.endpointName(JsonPath.stringAt("$.EndpointName"))
.endpointConfigName(JsonPath.stringAt("$.EndpointConfigName"))
.build();
csharp
new SageMakerCreateEndpoint(this, "SagemakerEndpoint", new SageMakerCreateEndpointProps {
EndpointName = JsonPath.StringAt("$.EndpointName"),
EndpointConfigName = JsonPath.StringAt("$.EndpointConfigName")
});
go
tasks.NewSageMakerCreateEndpoint(this, jsii.String("SagemakerEndpoint"), &SageMakerCreateEndpointProps{
EndpointName: sfn.JsonPath_StringAt(jsii.String("$.EndpointName")),
EndpointConfigName: sfn.JsonPath_*StringAt(jsii.String("$.EndpointConfigName")),
})
Create Endpoint Config
You can call the CreateEndpointConfig API from a Task state.
ts
new tasks.SageMakerCreateEndpointConfig(this, 'SagemakerEndpointConfig', {
endpointConfigName: 'MyEndpointConfig',
productionVariants: [{
initialInstanceCount: 2,
instanceType: ec2.InstanceType.of(ec2.InstanceClass.M5, ec2.InstanceSize.XLARGE),
modelName: 'MyModel',
variantName: 'awesome-variant',
}],
});
python
tasks.SageMakerCreateEndpointConfig(self, "SagemakerEndpointConfig",
endpoint_config_name="MyEndpointConfig",
production_variants=[tasks.ProductionVariant(
initial_instance_count=2,
instance_type=ec2.InstanceType.of(ec2.InstanceClass.M5, ec2.InstanceSize.XLARGE),
model_name="MyModel",
variant_name="awesome-variant"
)]
)
java
SageMakerCreateEndpointConfig.Builder.create(this, "SagemakerEndpointConfig")
.endpointConfigName("MyEndpointConfig")
.productionVariants(List.of(ProductionVariant.builder()
.initialInstanceCount(2)
.instanceType(InstanceType.of(InstanceClass.M5, InstanceSize.XLARGE))
.modelName("MyModel")
.variantName("awesome-variant")
.build()))
.build();
csharp
new SageMakerCreateEndpointConfig(this, "SagemakerEndpointConfig", new SageMakerCreateEndpointConfigProps {
EndpointConfigName = "MyEndpointConfig",
ProductionVariants = new [] { new ProductionVariant {
InitialInstanceCount = 2,
InstanceType = InstanceType.Of(InstanceClass.M5, InstanceSize.XLARGE),
ModelName = "MyModel",
VariantName = "awesome-variant"
} }
});
go
tasks.NewSageMakerCreateEndpointConfig(this, jsii.String("SagemakerEndpointConfig"), &SageMakerCreateEndpointConfigProps{
EndpointConfigName: jsii.String("MyEndpointConfig"),
ProductionVariants: []ProductionVariant{
&ProductionVariant{
InitialInstanceCount: jsii.Number(2),
InstanceType: ec2.InstanceType_Of(ec2.InstanceClass_M5, ec2.InstanceSize_XLARGE),
ModelName: jsii.String("MyModel"),
VariantName: jsii.String("awesome-variant"),
},
},
})
Create Model
You can call the CreateModel API from a Task state.
ts
new tasks.SageMakerCreateModel(this, 'Sagemaker', {
modelName: 'MyModel',
primaryContainer: new tasks.ContainerDefinition({
image: tasks.DockerImage.fromJsonExpression(sfn.JsonPath.stringAt('$.Model.imageName')),
mode: tasks.Mode.SINGLE_MODEL,
modelS3Location: tasks.S3Location.fromJsonExpression('$.TrainingJob.ModelArtifacts.S3ModelArtifacts'),
}),
});
python
tasks.SageMakerCreateModel(self, "Sagemaker",
model_name="MyModel",
primary_container=tasks.ContainerDefinition(
image=tasks.DockerImage.from_json_expression(sfn.JsonPath.string_at("$.Model.imageName")),
mode=tasks.Mode.SINGLE_MODEL,
model_s3_location=tasks.S3Location.from_json_expression("$.TrainingJob.ModelArtifacts.S3ModelArtifacts")
)
)
java
SageMakerCreateModel.Builder.create(this, "Sagemaker")
.modelName("MyModel")
.primaryContainer(ContainerDefinition.Builder.create()
.image(DockerImage.fromJsonExpression(JsonPath.stringAt("$.Model.imageName")))
.mode(Mode.SINGLE_MODEL)
.modelS3Location(S3Location.fromJsonExpression("$.TrainingJob.ModelArtifacts.S3ModelArtifacts"))
.build())
.build();
csharp
new SageMakerCreateModel(this, "Sagemaker", new SageMakerCreateModelProps {
ModelName = "MyModel",
PrimaryContainer = new ContainerDefinition(new ContainerDefinitionOptions {
Image = DockerImage.FromJsonExpression(JsonPath.StringAt("$.Model.imageName")),
Mode = Mode.SINGLE_MODEL,
ModelS3Location = S3Location.FromJsonExpression("$.TrainingJob.ModelArtifacts.S3ModelArtifacts")
})
});
go
tasks.NewSageMakerCreateModel(this, jsii.String("Sagemaker"), &SageMakerCreateModelProps{
ModelName: jsii.String("MyModel"),
PrimaryContainer: tasks.NewContainerDefinition(&ContainerDefinitionOptions{
Image: tasks.DockerImage_FromJsonExpression(sfn.JsonPath_StringAt(jsii.String("$.Model.imageName"))),
Mode: tasks.Mode_SINGLE_MODEL,
ModelS3Location: tasks.S3Location_FromJsonExpression(jsii.String("$.TrainingJob.ModelArtifacts.S3ModelArtifacts")),
}),
})
Update Endpoint
You can call the UpdateEndpoint API from a Task state.
ts
new tasks.SageMakerUpdateEndpoint(this, 'SagemakerEndpoint', {
endpointName: sfn.JsonPath.stringAt('$.Endpoint.Name'),
endpointConfigName: sfn.JsonPath.stringAt('$.Endpoint.EndpointConfig'),
});
python
tasks.SageMakerUpdateEndpoint(self, "SagemakerEndpoint",
endpoint_name=sfn.JsonPath.string_at("$.Endpoint.Name"),
endpoint_config_name=sfn.JsonPath.string_at("$.Endpoint.EndpointConfig")
)
java
SageMakerUpdateEndpoint.Builder.create(this, "SagemakerEndpoint")
.endpointName(JsonPath.stringAt("$.Endpoint.Name"))
.endpointConfigName(JsonPath.stringAt("$.Endpoint.EndpointConfig"))
.build();
csharp
new SageMakerUpdateEndpoint(this, "SagemakerEndpoint", new SageMakerUpdateEndpointProps {
EndpointName = JsonPath.StringAt("$.Endpoint.Name"),
EndpointConfigName = JsonPath.StringAt("$.Endpoint.EndpointConfig")
});
go
tasks.NewSageMakerUpdateEndpoint(this, jsii.String("SagemakerEndpoint"), &SageMakerUpdateEndpointProps{
EndpointName: sfn.JsonPath_StringAt(jsii.String("$.Endpoint.Name")),
EndpointConfigName: sfn.JsonPath_*StringAt(jsii.String("$.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.
ts
const topic = new sns.Topic(this, 'Topic');
// Use a field from the execution data as message.
const task1 = new tasks.SnsPublish(this, 'Publish1', {
topic,
integrationPattern: sfn.IntegrationPattern.REQUEST_RESPONSE,
message: sfn.TaskInput.fromDataAt('$.state.message'),
messageAttributes: {
place: {
value: sfn.JsonPath.stringAt('$.place'),
},
pic: {
// BINARY must be explicitly set
dataType: tasks.MessageAttributeDataType.BINARY,
value: sfn.JsonPath.stringAt('$.pic'),
},
people: {
value: 4,
},
handles: {
value: ['@kslater', '@jjf', null, '@mfanning'],
},
},
});
// Combine a field from the execution data with
// a literal object.
const task2 = new tasks.SnsPublish(this, 'Publish2', {
topic,
message: sfn.TaskInput.fromObject({
field1: 'somedata',
field2: sfn.JsonPath.stringAt('$.field2'),
}),
});
python
topic = sns.Topic(self, "Topic")
# Use a field from the execution data as message.
task1 = tasks.SnsPublish(self, "Publish1",
topic=topic,
integration_pattern=sfn.IntegrationPattern.REQUEST_RESPONSE,
message=sfn.TaskInput.from_data_at("$.state.message"),
message_attributes={
"place": tasks.MessageAttribute(
value=sfn.JsonPath.string_at("$.place")
),
"pic": tasks.MessageAttribute(
# BINARY must be explicitly set
data_type=tasks.MessageAttributeDataType.BINARY,
value=sfn.JsonPath.string_at("$.pic")
),
"people": tasks.MessageAttribute(
value=4
),
"handles": tasks.MessageAttribute(
value=["@kslater", "@jjf", null, "@mfanning"]
)
}
)
# Combine a field from the execution data with
# a literal object.
task2 = tasks.SnsPublish(self, "Publish2",
topic=topic,
message=sfn.TaskInput.from_object({
"field1": "somedata",
"field2": sfn.JsonPath.string_at("$.field2")
})
)
java
Topic topic = new Topic(this, "Topic");
// Use a field from the execution data as message.
SnsPublish task1 = SnsPublish.Builder.create(this, "Publish1")
.topic(topic)
.integrationPattern(IntegrationPattern.REQUEST_RESPONSE)
.message(TaskInput.fromDataAt("$.state.message"))
.messageAttributes(Map.of(
"place", MessageAttribute.builder()
.value(JsonPath.stringAt("$.place"))
.build(),
"pic", MessageAttribute.builder()
// BINARY must be explicitly set
.dataType(MessageAttributeDataType.BINARY)
.value(JsonPath.stringAt("$.pic"))
.build(),
"people", MessageAttribute.builder()
.value(4)
.build(),
"handles", MessageAttribute.builder()
.value(List.of("@kslater", "@jjf", null, "@mfanning"))
.build()))
.build();
// Combine a field from the execution data with
// a literal object.
SnsPublish task2 = SnsPublish.Builder.create(this, "Publish2")
.topic(topic)
.message(TaskInput.fromObject(Map.of(
"field1", "somedata",
"field2", JsonPath.stringAt("$.field2"))))
.build();
csharp
var topic = new Topic(this, "Topic");
// Use a field from the execution data as message.
var task1 = new SnsPublish(this, "Publish1", new SnsPublishProps {
Topic = topic,
IntegrationPattern = IntegrationPattern.REQUEST_RESPONSE,
Message = TaskInput.FromDataAt("$.state.message"),
MessageAttributes = new Dictionary<string, MessageAttribute> {
{ "place", new MessageAttribute {
Value = JsonPath.StringAt("$.place")
} },
{ "pic", new MessageAttribute {
// BINARY must be explicitly set
DataType = MessageAttributeDataType.BINARY,
Value = JsonPath.StringAt("$.pic")
} },
{ "people", new MessageAttribute {
Value = 4
} },
{ "handles", new MessageAttribute {
Value = new [] { "@kslater", "@jjf", null, "@mfanning" }
} }
}
});
// Combine a field from the execution data with
// a literal object.
var task2 = new SnsPublish(this, "Publish2", new SnsPublishProps {
Topic = topic,
Message = TaskInput.FromObject(new Dictionary<string, object> {
{ "field1", "somedata" },
{ "field2", JsonPath.StringAt("$.field2") }
})
});
go
topic := sns.NewTopic(this, jsii.String("Topic"))
// Use a field from the execution data as message.
task1 := tasks.NewSnsPublish(this, jsii.String("Publish1"), &SnsPublishProps{
Topic: Topic,
IntegrationPattern: sfn.IntegrationPattern_REQUEST_RESPONSE,
Message: sfn.TaskInput_FromDataAt(jsii.String("$.state.message")),
MessageAttributes: map[string]MessageAttribute{
"place": &MessageAttribute{
"value": sfn.JsonPath_stringAt(jsii.String("$.place")),
},
"pic": &MessageAttribute{
// BINARY must be explicitly set
"dataType": tasks.MessageAttributeDataType_BINARY,
"value": sfn.JsonPath_stringAt(jsii.String("$.pic")),
},
"people": &MessageAttribute{
"value": jsii.Number(4),
},
"handles": &MessageAttribute{
"value": []interface{}{
jsii.String("@kslater"),
jsii.String("@jjf"),
nil,
jsii.String("@mfanning"),
},
},
},
})
// Combine a field from the execution data with
// a literal object.
task2 := tasks.NewSnsPublish(this, jsii.String("Publish2"), &SnsPublishProps{
Topic: Topic,
Message: sfn.TaskInput_FromObject(map[string]interface{}{
"field1": jsii.String("somedata"),
"field2": sfn.JsonPath_stringAt(jsii.String("$.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.
ts
// Define a state machine with one Pass state
const child = new sfn.StateMachine(this, 'ChildStateMachine', {
definition: sfn.Chain.start(new sfn.Pass(this, 'PassState')),
});
// Include the state machine in a Task state with callback pattern
const task = new tasks.StepFunctionsStartExecution(this, 'ChildTask', {
stateMachine: child,
integrationPattern: sfn.IntegrationPattern.WAIT_FOR_TASK_TOKEN,
input: sfn.TaskInput.fromObject({
token: sfn.JsonPath.taskToken,
foo: 'bar',
}),
name: 'MyExecutionName',
});
// Define a second state machine with the Task state above
new sfn.StateMachine(this, 'ParentStateMachine', {
definition: task,
});
python
# Define a state machine with one Pass state
child = sfn.StateMachine(self, "ChildStateMachine",
definition=sfn.Chain.start(sfn.Pass(self, "PassState"))
)
# Include the state machine in a Task state with callback pattern
task = tasks.StepFunctionsStartExecution(self, "ChildTask",
state_machine=child,
integration_pattern=sfn.IntegrationPattern.WAIT_FOR_TASK_TOKEN,
input=sfn.TaskInput.from_object({
"token": sfn.JsonPath.task_token,
"foo": "bar"
}),
name="MyExecutionName"
)
# Define a second state machine with the Task state above
sfn.StateMachine(self, "ParentStateMachine",
definition=task
)
java
// Define a state machine with one Pass state
StateMachine child = StateMachine.Builder.create(this, "ChildStateMachine")
.definition(Chain.start(new Pass(this, "PassState")))
.build();
// Include the state machine in a Task state with callback pattern
StepFunctionsStartExecution task = StepFunctionsStartExecution.Builder.create(this, "ChildTask")
.stateMachine(child)
.integrationPattern(IntegrationPattern.WAIT_FOR_TASK_TOKEN)
.input(TaskInput.fromObject(Map.of(
"token", JsonPath.getTaskToken(),
"foo", "bar")))
.name("MyExecutionName")
.build();
// Define a second state machine with the Task state above
// Define a second state machine with the Task state above
StateMachine.Builder.create(this, "ParentStateMachine")
.definition(task)
.build();
csharp
// Define a state machine with one Pass state
var child = new StateMachine(this, "ChildStateMachine", new StateMachineProps {
Definition = Chain.Start(new Pass(this, "PassState"))
});
// Include the state machine in a Task state with callback pattern
var task = new StepFunctionsStartExecution(this, "ChildTask", new StepFunctionsStartExecutionProps {
StateMachine = child,
IntegrationPattern = IntegrationPattern.WAIT_FOR_TASK_TOKEN,
Input = TaskInput.FromObject(new Dictionary<string, object> {
{ "token", JsonPath.TaskToken },
{ "foo", "bar" }
}),
Name = "MyExecutionName"
});
// Define a second state machine with the Task state above
// Define a second state machine with the Task state above
new StateMachine(this, "ParentStateMachine", new StateMachineProps {
Definition = task
});
go
// Define a state machine with one Pass state
child := sfn.NewStateMachine(this, jsii.String("ChildStateMachine"), &StateMachineProps{
Definition: sfn.Chain_Start(sfn.NewPass(this, jsii.String("PassState"))),
})
// Include the state machine in a Task state with callback pattern
task := tasks.NewStepFunctionsStartExecution(this, jsii.String("ChildTask"), &StepFunctionsStartExecutionProps{
StateMachine: child,
IntegrationPattern: sfn.IntegrationPattern_WAIT_FOR_TASK_TOKEN,
Input: sfn.TaskInput_FromObject(map[string]interface{}{
"token": sfn.JsonPath_taskToken(),
"foo": jsii.String("bar"),
}),
Name: jsii.String("MyExecutionName"),
})
// Define a second state machine with the Task state above
// Define a second state machine with the Task state above
sfn.NewStateMachine(this, jsii.String("ParentStateMachine"), &StateMachineProps{
Definition: task,
})
You can utilize Associate Workflow Executions via the associateWithParent property. This allows the Step Functions UI to link child executions from parent executions, making it easier to trace execution flow across state machines.
ts
declare const child: sfn.StateMachine;
const task = new tasks.StepFunctionsStartExecution(this, 'ChildTask', {
stateMachine: child,
associateWithParent: true,
});
python
# child: sfn.StateMachine
task = tasks.StepFunctionsStartExecution(self, "ChildTask",
state_machine=child,
associate_with_parent=True
)
java
StateMachine child;
StepFunctionsStartExecution task = StepFunctionsStartExecution.Builder.create(this, "ChildTask")
.stateMachine(child)
.associateWithParent(true)
.build();
csharp
StateMachine child;
var task = new StepFunctionsStartExecution(this, "ChildTask", new StepFunctionsStartExecutionProps {
StateMachine = child,
AssociateWithParent = true
});
go
var child StateMachine
task := tasks.NewStepFunctionsStartExecution(this, jsii.String("ChildTask"), &StepFunctionsStartExecutionProps{
StateMachine: child,
AssociateWithParent: jsii.Boolean(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.
ts
const submitJobActivity = new sfn.Activity(this, 'SubmitJob');
new tasks.StepFunctionsInvokeActivity(this, 'Submit Job', {
activity: submitJobActivity,
});
python
submit_job_activity = sfn.Activity(self, "SubmitJob")
tasks.StepFunctionsInvokeActivity(self, "Submit Job",
activity=submit_job_activity
)
java
Activity submitJobActivity = new Activity(this, "SubmitJob");
StepFunctionsInvokeActivity.Builder.create(this, "Submit Job")
.activity(submitJobActivity)
.build();
csharp
var submitJobActivity = new Activity(this, "SubmitJob");
new StepFunctionsInvokeActivity(this, "Submit Job", new StepFunctionsInvokeActivityProps {
Activity = submitJobActivity
});
go
submitJobActivity := sfn.NewActivity(this, jsii.String("SubmitJob"))
tasks.NewStepFunctionsInvokeActivity(this, jsii.String("Submit Job"), &StepFunctionsInvokeActivityProps{
Activity: submitJobActivity,
})
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.
ts
const submitJobActivity = new sfn.Activity(this, 'SubmitJob');
new tasks.StepFunctionsInvokeActivity(this, 'Submit Job', {
activity: submitJobActivity,
parameters: {
comment: 'Selecting what I care about.',
MyDetails: {
size: sfn.JsonPath.stringAt('$.product.details.size'),
exists: sfn.JsonPath.stringAt('$.product.availability'),
StaticValue: 'foo'
},
},
});
python
submit_job_activity = sfn.Activity(self, "SubmitJob")
tasks.StepFunctionsInvokeActivity(self, "Submit Job",
activity=submit_job_activity,
parameters={
"comment": "Selecting what I care about.",
"MyDetails": {
"size": sfn.JsonPath.string_at("$.product.details.size"),
"exists": sfn.JsonPath.string_at("$.product.availability"),
"StaticValue": "foo"
}
}
)
java
Activity submitJobActivity = new Activity(this, "SubmitJob");
StepFunctionsInvokeActivity.Builder.create(this, "Submit Job")
.activity(submitJobActivity)
.parameters(Map.of(
"comment", "Selecting what I care about.",
"MyDetails", Map.of(
"size", JsonPath.stringAt("$.product.details.size"),
"exists", JsonPath.stringAt("$.product.availability"),
"StaticValue", "foo")))
.build();
csharp
var submitJobActivity = new Activity(this, "SubmitJob");
new StepFunctionsInvokeActivity(this, "Submit Job", new StepFunctionsInvokeActivityProps {
Activity = submitJobActivity,
Parameters = new Dictionary<string, object> {
{ "comment", "Selecting what I care about." },
{ "MyDetails", new Dictionary<string, object> {
{ "size", JsonPath.StringAt("$.product.details.size") },
{ "exists", JsonPath.StringAt("$.product.availability") },
{ "StaticValue", "foo" }
} }
}
});
go
submitJobActivity := sfn.NewActivity(this, jsii.String("SubmitJob"))
tasks.NewStepFunctionsInvokeActivity(this, jsii.String("Submit Job"), &StepFunctionsInvokeActivityProps{
Activity: submitJobActivity,
Parameters: map[string]interface{}{
"comment": jsii.String("Selecting what I care about."),
"MyDetails": map[string]interface{}{
"size": sfn.JsonPath_stringAt(jsii.String("$.product.details.size")),
"exists": sfn.JsonPath_stringAt(jsii.String("$.product.availability")),
"StaticValue": jsii.String("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.
ts
const queue = new sqs.Queue(this, 'Queue');
// Use a field from the execution data as message.
const task1 = new tasks.SqsSendMessage(this, 'Send1', {
queue,
messageBody: sfn.TaskInput.fromJsonPathAt('$.message'),
});
// Combine a field from the execution data with
// a literal object.
const task2 = new tasks.SqsSendMessage(this, 'Send2', {
queue,
messageBody: sfn.TaskInput.fromObject({
field1: 'somedata',
field2: sfn.JsonPath.stringAt('$.field2'),
}),
});
python
queue = sqs.Queue(self, "Queue")
# Use a field from the execution data as message.
task1 = tasks.SqsSendMessage(self, "Send1",
queue=queue,
message_body=sfn.TaskInput.from_json_path_at("$.message")
)
# Combine a field from the execution data with
# a literal object.
task2 = tasks.SqsSendMessage(self, "Send2",
queue=queue,
message_body=sfn.TaskInput.from_object({
"field1": "somedata",
"field2": sfn.JsonPath.string_at("$.field2")
})
)
java
Queue queue = new Queue(this, "Queue");
// Use a field from the execution data as message.
SqsSendMessage task1 = SqsSendMessage.Builder.create(this, "Send1")
.queue(queue)
.messageBody(TaskInput.fromJsonPathAt("$.message"))
.build();
// Combine a field from the execution data with
// a literal object.
SqsSendMessage task2 = SqsSendMessage.Builder.create(this, "Send2")
.queue(queue)
.messageBody(TaskInput.fromObject(Map.of(
"field1", "somedata",
"field2", JsonPath.stringAt("$.field2"))))
.build();
csharp
var queue = new Queue(this, "Queue");
// Use a field from the execution data as message.
var task1 = new SqsSendMessage(this, "Send1", new SqsSendMessageProps {
Queue = queue,
MessageBody = TaskInput.FromJsonPathAt("$.message")
});
// Combine a field from the execution data with
// a literal object.
var task2 = new SqsSendMessage(this, "Send2", new SqsSendMessageProps {
Queue = queue,
MessageBody = TaskInput.FromObject(new Dictionary<string, object> {
{ "field1", "somedata" },
{ "field2", JsonPath.StringAt("$.field2") }
})
});
go
queue := sqs.NewQueue(this, jsii.String("Queue"))
// Use a field from the execution data as message.
task1 := tasks.NewSqsSendMessage(this, jsii.String("Send1"), &SqsSendMessageProps{
Queue: Queue,
MessageBody: sfn.TaskInput_FromJsonPathAt(jsii.String("$.message")),
})
// Combine a field from the execution data with
// a literal object.
task2 := tasks.NewSqsSendMessage(this, jsii.String("Send2"), &SqsSendMessageProps{
Queue: Queue,
MessageBody: sfn.TaskInput_FromObject(map[string]interface{}{
"field1": jsii.String("somedata"),
"field2": sfn.JsonPath_stringAt(jsii.String("$.field2")),
}),
})
Types
| Name | Kind | Summary |
|---|---|---|
AcceleratorClass | Class | The generation of Elastic Inference (EI) instance. |
AcceleratorType | Class | The size of the Elastic Inference (EI) instance to use for the production variant. |
ActionAfterCompletion | Enum | The action that EventBridge Scheduler applies to the schedule after the schedule completes invoking the target. |
ActionOnFailure | Enum | The action to take when the cluster step fails. |
AlgorithmSpecification | Struct | Specify the training algorithm and algorithm-specific metadata. |
ApplicationConfiguration | Struct | A configuration specification to be used when provisioning virtual clusters, which can include configurations for applications and software bundled with Amazon EMR on EKS. |
AssembleWith | Enum | How to assemble the results of the transform job as a single S3 object. |
AthenaGetQueryExecution | Construct | Get an Athena Query Execution as a Task. |
AthenaGetQueryExecutionJsonataProps | Struct | Properties for getting a Query Execution using JSONata. |
AthenaGetQueryExecutionJsonPathProps | Struct | Properties for getting a Query Execution using JSONPath. |
AthenaGetQueryExecutionProps | Struct | Properties for getting a Query Execution. |
AthenaGetQueryResults | Construct | Get an Athena Query Results as a Task. |
AthenaGetQueryResultsJsonataProps | Struct | Properties for getting a Query Results using JSONata. |
AthenaGetQueryResultsJsonPathProps | Struct | Properties for getting a Query Results using JSONPath. |
AthenaGetQueryResultsProps | Struct | Properties for getting a Query Results. |
AthenaStartQueryExecution | Construct | Start an Athena Query as a Task. |
AthenaStartQueryExecutionJsonataProps | Struct | Properties for starting a Query Execution using JSONata. |
AthenaStartQueryExecutionJsonPathProps | Struct | Properties for starting a Query Execution using JSONPath. |
AthenaStartQueryExecutionProps | Struct | Properties for starting a Query Execution. |
AthenaStopQueryExecution | Construct | Stop an Athena Query Execution as a Task. |
AthenaStopQueryExecutionJsonataProps | Struct | Properties for stopping a Query Execution using JSONata. |
AthenaStopQueryExecutionJsonPathProps | Struct | Properties for stopping a Query Execution using JSONPath. |
AthenaStopQueryExecutionProps | Struct | Properties for stopping a Query Execution. |
AuthType | Enum | The authentication method used to call the endpoint. |
BatchContainerOverrides | Struct | The overrides that should be sent to a container. |
BatchJobDependency | Struct | An object representing an AWS Batch job dependency. |
BatchStrategy | Enum | Specifies the number of records to include in a mini-batch for an HTTP inference request. |
BatchSubmitJob | Construct | Task to submits an AWS Batch job from a job definition. |
BatchSubmitJobJsonataProps | Struct | Properties for BatchSubmitJob using JSONata. |
BatchSubmitJobJsonPathProps | Struct | Properties for BatchSubmitJob using JSONPath. |
BatchSubmitJobProps | Struct | Properties for BatchSubmitJob. |
BedrockCreateModelCustomizationJob | Construct | A Step Functions Task to create model customization in Bedrock. |
BedrockCreateModelCustomizationJobProps | Struct | Properties for invoking a Bedrock Model. |
BedrockInvokeModel | Construct | A Step Functions Task to invoke a model in Bedrock. |
BedrockInvokeModelInputProps | Struct | Location to retrieve the input data, prior to calling Bedrock InvokeModel. |
BedrockInvokeModelJsonataProps | Struct | Properties for invoking a Bedrock Model. |
BedrockInvokeModelJsonPathProps | Struct | Properties for invoking a Bedrock Model. |
BedrockInvokeModelOutputProps | Struct | Location where the Bedrock InvokeModel API response is written. |
BedrockInvokeModelProps | Struct | Properties for invoking a Bedrock Model. |
CallApiGatewayEndpointBaseOptions | Struct | Base CallApiGatewayEdnpoint Task Props. |
CallApiGatewayEndpointBaseProps | Struct | Base CallApiGatewayEndpoint Task Props. |
CallApiGatewayEndpointJsonataBaseProps | Struct | Base CallApiGatewayEndpoint Task Props. |
CallApiGatewayEndpointJsonPathBaseProps | Struct | Base CallApiGatewayEndpoint Task Props. |
CallApiGatewayHttpApiEndpoint | Construct | Call HTTP API endpoint as a Task. |
CallApiGatewayHttpApiEndpointJsonataProps | Struct | Properties for calling an HTTP API Endpoint using JSONata. |
CallApiGatewayHttpApiEndpointJsonPathProps | Struct | Properties for calling an HTTP API Endpoint using JSONPath. |
CallApiGatewayHttpApiEndpointOptions | Struct | Base properties for calling an HTTP API Endpoint. |
CallApiGatewayHttpApiEndpointProps | Struct | Properties for calling an HTTP API Endpoint. |
CallApiGatewayRestApiEndpoint | Construct | Call REST API endpoint as a Task. |
CallApiGatewayRestApiEndpointJsonataProps | Struct | Properties for calling an REST API Endpoint using JSONata. |
CallApiGatewayRestApiEndpointJsonPathProps | Struct | Properties for calling an REST API Endpoint using JSONPath. |
CallApiGatewayRestApiEndpointOptions | Struct | Base properties for calling an REST API Endpoint. |
CallApiGatewayRestApiEndpointProps | Struct | Properties for calling an REST API Endpoint. |
CallAwsService | Construct | A StepFunctions task to call an AWS service API. |
CallAwsServiceCrossRegion | Construct | A Step Functions task to call an AWS service API across regions. |
CallAwsServiceCrossRegionJsonataProps | Struct | Properties for calling an AWS service's API action using JSONata from your state machine across regions. |
CallAwsServiceCrossRegionJsonPathProps | Struct | Properties for calling an AWS service's API action using JSONPath from your state machine across regions. |
CallAwsServiceCrossRegionProps | Struct | Properties for calling an AWS service's API action from your state machine across regions. |
CallAwsServiceJsonataProps | Struct | Properties for calling an AWS service's API action using JSONata from your state machine. |
CallAwsServiceJsonPathProps | Struct | Properties for calling an AWS service's API action using JSONPath from your state machine. |
CallAwsServiceProps | Struct | Properties for calling an AWS service's API action from your state machine. |
CapacityProviderOptions | Class | Capacity provider options. |
Channel | Struct | Describes the training, validation or test dataset and the Amazon S3 location where it is stored. |
Classification | Class | The classification within a EMR Containers application configuration. |
CodeBuildStartBuild | Construct | Start a CodeBuild Build as a task. |
CodeBuildStartBuildBatch | Construct | Start a CodeBuild BatchBuild as a task. |
CodeBuildStartBuildBatchJsonataProps | Struct | Properties for CodeBuildStartBuildBatch using JSONata. |
CodeBuildStartBuildBatchJsonPathProps | Struct | Properties for CodeBuildStartBuildBatch using JSONPath. |
CodeBuildStartBuildBatchProps | Struct | Properties for CodeBuildStartBuildBatch. |
CodeBuildStartBuildJsonataProps | Struct | Properties for CodeBuildStartBuild using JSONata. |
CodeBuildStartBuildJsonPathProps | Struct | Properties for CodeBuildStartBuild using JSONPath. |
CodeBuildStartBuildProps | Struct | Properties for CodeBuildStartBuild. |
CommonEcsRunTaskProps | Struct | Basic properties for ECS Tasks. |
CompressionType | Enum | Compression type of the data. |
ContainerDefinition | Class | Describes the container, as part of model definition. |
ContainerDefinitionConfig | Struct | Configuration options for the ContainerDefinition. |
ContainerDefinitionOptions | Struct | Properties to define a ContainerDefinition. |
ContainerOverride | Struct | A list of container overrides that specify the name of a container and the overrides it should receive. |
ContainerOverrides | Struct | The overrides that should be sent to a container. |
CronOptions | Struct | Options to configure a cron expression. |
CustomizationType | Enum | The customization type. |
CustomModelTag | Struct | The key/value pair for a tag. |
DataBucketConfiguration | Struct | S3 bucket configuration for data storage destination. |
DataSource | Struct | Location of the channel data. |
DockerImage | Class | Creates IDockerImage instances. |
DockerImageConfig | Struct | Configuration for a using Docker image. |
DynamoAttributeValue | Class | Represents the data for an attribute. |
DynamoConsumedCapacity | Enum | Determines the level of detail about provisioned throughput consumption that is returned. |
DynamoDeleteItem | Construct | A StepFunctions task to call DynamoDeleteItem. |
DynamoDeleteItemJsonataProps | Struct | Properties for DynamoDeleteItem Task using JSONata. |
DynamoDeleteItemJsonPathProps | Struct | Properties for DynamoDeleteItem Task using JSONPath. |
DynamoDeleteItemProps | Struct | Properties for DynamoDeleteItem Task. |
DynamoGetItem | Construct | A StepFunctions task to call DynamoGetItem. |
DynamoGetItemJsonataProps | Struct | Properties for DynamoGetItem Task using JSONata. |
DynamoGetItemJsonPathProps | Struct | Properties for DynamoGetItem Task using JSONPath. |
DynamoGetItemProps | Struct | Properties for DynamoGetItem Task. |
DynamoItemCollectionMetrics | Enum | Determines whether item collection metrics are returned. |
DynamoProjectionExpression | Class | Class to generate projection expression. |
DynamoPutItem | Construct | A StepFunctions task to call DynamoPutItem. |
DynamoPutItemJsonataProps | Struct | Properties for DynamoPutItem Task using JSONata. |
DynamoPutItemJsonPathProps | Struct | Properties for DynamoPutItem Task using JSONPath. |
DynamoPutItemProps | Struct | Properties for DynamoPutItem Task. |
DynamoReturnValues | Enum | Use ReturnValues if you want to get the item attributes as they appear before or after they are changed. |
DynamoUpdateItem | Construct | A StepFunctions task to call DynamoUpdateItem. |
DynamoUpdateItemJsonataProps | Struct | Properties for DynamoUpdateItem Task using JSONata. |
DynamoUpdateItemJsonPathProps | Struct | Properties for DynamoUpdateItem Task using JSONPath. |
DynamoUpdateItemProps | Struct | Properties for DynamoUpdateItem Task. |
EcsEc2LaunchTarget | Class | Configuration for running an ECS task on EC2. |
EcsEc2LaunchTargetOptions | Struct | Options to run an ECS task on EC2 in StepFunctions and ECS. |
EcsFargateLaunchTarget | Class | Configuration for running an ECS task on Fargate. |
EcsFargateLaunchTargetOptions | Struct | Properties to define an ECS service. |
EcsLaunchTargetConfig | Struct | Configuration options for the ECS launch type. |
EcsRunTask | Construct | Run a Task on ECS or Fargate. |
EcsRunTaskJsonataProps | Struct | Properties for ECS Tasks using JSONata. |
EcsRunTaskJsonPathProps | Struct | Properties for ECS Tasks using JSONPath. |
EcsRunTaskProps | Struct | Properties for ECS Tasks. |
EksCall | Construct | Call a EKS endpoint as a Task. |
EksCallJsonataProps | Struct | Properties for calling a EKS endpoint with EksCall using JSONata. |
EksCallJsonPathProps | Struct | Properties for calling a EKS endpoint with EksCall using JSONPath. |
EksCallProps | Struct | Properties for calling a EKS endpoint with EksCall. |
EksClusterInput | Class | Class that supports methods which return the EKS cluster name depending on input type. |
EmrAddStep | Construct | A Step Functions Task to add a Step to an EMR Cluster. |
EmrAddStepJsonataProps | Struct | Properties for EmrAddStep using JSONata. |
EmrAddStepJsonPathProps | Struct | Properties for EmrAddStep using JSONPath. |
EmrAddStepProps | Struct | Properties for EmrAddStep. |
EmrCancelStep | Construct | A Step Functions task to cancel a Step on an EMR Cluster. |
EmrCancelStepJsonataProps | Struct | Properties for calling an EMR CancelStep using JSONata from your state machine. |
EmrCancelStepJsonPathProps | Struct | Properties for calling an EMR CancelStep using JSONPath from your state machine. |
EmrCancelStepProps | Struct | Properties for calling an EMR CancelStep from your state machine. |
EmrContainersCreateVirtualCluster | Construct | Task that creates an EMR Containers virtual cluster from an EKS cluster. |
EmrContainersCreateVirtualClusterJsonataProps | Struct | Properties to define a EMR Containers CreateVirtualCluster Task using JSONata on an EKS cluster. |
EmrContainersCreateVirtualClusterJsonPathProps | Struct | Properties to define a EMR Containers CreateVirtualCluster Task using JSONPath on an EKS cluster. |
EmrContainersCreateVirtualClusterProps | Struct | Properties to define a EMR Containers CreateVirtualCluster Task on an EKS cluster. |
EmrContainersDeleteVirtualCluster | Construct | Deletes an EMR Containers virtual cluster as a Task. |
EmrContainersDeleteVirtualClusterJsonataProps | Struct | Properties to define a EMR Containers DeleteVirtualCluster Task using JSONata. |
EmrContainersDeleteVirtualClusterJsonPathProps | Struct | Properties to define a EMR Containers DeleteVirtualCluster Task using JSONPath. |
EmrContainersDeleteVirtualClusterProps | Struct | Properties to define a EMR Containers DeleteVirtualCluster Task. |
EmrContainersStartJobRun | Construct | Starts a job run. |
EmrContainersStartJobRunJsonataProps | Struct | Properties for calling EMR Containers StartJobRun using JSONata. |
EmrContainersStartJobRunJsonPathProps | Struct | Properties for calling EMR Containers StartJobRun using JSONPath. |
EmrContainersStartJobRunProps | Struct | The props for a EMR Containers StartJobRun Task. |
EmrCreateCluster | Construct | A Step Functions Task to create an EMR Cluster. |
EmrCreateCluster.ApplicationConfigProperty | Struct | Properties for the EMR Cluster Applications. |
EmrCreateCluster.AutoScalingPolicyProperty | Struct | An automatic scaling policy for a core instance group or task instance group in an Amazon EMR cluster. |
EmrCreateCluster.BootstrapActionConfigProperty | Struct | Configuration of a bootstrap action. |
EmrCreateCluster.CloudWatchAlarmComparisonOperator | Enum | CloudWatch Alarm Comparison Operators. |
EmrCreateCluster.CloudWatchAlarmDefinitionProperty | Struct | The definition of a CloudWatch metric alarm, which determines when an automatic scaling activity is triggered. |
EmrCreateCluster.CloudWatchAlarmStatistic | Enum | CloudWatch Alarm Statistics. |
EmrCreateCluster.CloudWatchAlarmUnit | Enum | CloudWatch Alarm Units. |
EmrCreateCluster.ComputeLimitsUnitType | Enum | The unit type for managed scaling policy compute limits. |
EmrCreateCluster.ConfigurationProperty | Struct | An optional configuration specification to be used when provisioning cluster instances, which can include configurations for applications and software bundled with Amazon EMR. |
EmrCreateCluster.EbsBlockDeviceConfigProperty | Struct | Configuration of requested EBS block device associated with the instance group with count of volumes that will be associated to every instance. |
EmrCreateCluster.EbsBlockDeviceVolumeType | Enum | EBS Volume Types. |
EmrCreateCluster.EbsConfigurationProperty | Struct | The Amazon EBS configuration of a cluster instance. |
EmrCreateCluster.EmrClusterScaleDownBehavior | Enum | The Cluster ScaleDownBehavior specifies the way that individual Amazon EC2 instances terminate when an automatic scale-in activity occurs or an instance group is resized. |
EmrCreateCluster.InstanceFleetConfigProperty | Struct | The configuration that defines an instance fleet. |
EmrCreateCluster.InstanceFleetProvisioningSpecificationsProperty | Struct | The launch specification for On-Demand and Spot instances in the fleet, which determines the defined duration and provisioning timeout behavior, and allocation strategy. |
EmrCreateCluster.InstanceGroupConfigProperty | Struct | Configuration defining a new instance group. |
EmrCreateCluster.InstanceMarket | Enum | EC2 Instance Market. |
EmrCreateCluster.InstanceRoleType | Enum | Instance Role Types. |
EmrCreateCluster.InstancesConfigProperty | Struct | A specification of the number and type of Amazon EC2 instances. |
EmrCreateCluster.InstanceTypeConfigProperty | Struct | An instance type configuration for each instance type in an instance fleet, which determines the EC2 instances Amazon EMR attempts to provision to fulfill On-Demand and Spot target capacities. |
EmrCreateCluster.KerberosAttributesProperty | Struct | Attributes for Kerberos configuration when Kerberos authentication is enabled using a security configuration. |
EmrCreateCluster.ManagedScalingComputeLimitsProperty | Struct | The EC2 unit limits for a managed scaling policy. |
EmrCreateCluster.ManagedScalingPolicyProperty | Struct | The managed scaling policy for an Amazon EMR cluster. |
EmrCreateCluster.MetricDimensionProperty | Struct | A CloudWatch dimension, which is specified using a Key (known as a Name in CloudWatch), Value pair. |
EmrCreateCluster.OnDemandAllocationStrategy | Enum | On-Demand Allocation Strategies. |
EmrCreateCluster.OnDemandProvisioningSpecificationProperty | Struct | The launch specification for On-Demand Instances in the instance fleet, which determines the allocation strategy. |
EmrCreateCluster.PlacementTypeProperty | Struct | The Amazon EC2 Availability Zone configuration of the cluster (job flow). |
EmrCreateCluster.ScalingActionProperty | Struct | The type of adjustment the automatic scaling activity makes when triggered, and the periodicity of the adjustment. |
EmrCreateCluster.ScalingAdjustmentType | Enum | AutoScaling Adjustment Type. |
EmrCreateCluster.ScalingConstraintsProperty | Struct | The upper and lower EC2 instance limits for an automatic scaling policy. |
EmrCreateCluster.ScalingRuleProperty | Struct | A scale-in or scale-out rule that defines scaling activity, including the CloudWatch metric alarm that triggers activity, how EC2 instances are added or removed, and the periodicity of adjustments. |
EmrCreateCluster.ScalingTriggerProperty | Struct | The conditions that trigger an automatic scaling activity and the definition of a CloudWatch metric alarm. |
EmrCreateCluster.ScriptBootstrapActionConfigProperty | Struct | Configuration of the script to run during a bootstrap action. |
EmrCreateCluster.SimpleScalingPolicyConfigurationProperty | Struct | An automatic scaling configuration, which describes how the policy adds or removes instances, the cooldown period, and the number of EC2 instances that will be added each time the CloudWatch metric alarm condition is satisfied. |
EmrCreateCluster.SpotAllocationStrategy | Enum | Spot Allocation Strategies. |
EmrCreateCluster.SpotProvisioningSpecificationProperty | Struct | The launch specification for Spot instances in the instance fleet, which determines the defined duration and provisioning timeout behavior. |
EmrCreateCluster.SpotTimeoutAction | Enum | Spot Timeout Actions. |
EmrCreateCluster.VolumeSpecificationProperty | Struct | EBS volume specifications such as volume type, IOPS, and size (GiB) that will be requested for the EBS volume attached to an EC2 instance in the cluster. |
EmrCreateClusterJsonataProps | Struct | Properties for calling an AWS service's API action using JSONata from your state machine across regions. |
EmrCreateClusterJsonPathProps | Struct | Properties for calling an AWS service's API action using JSONPath from your state machine across regions. |
EmrCreateClusterProps | Struct | Properties for calling an AWS service's API action from your state machine across regions. |
EmrModifyInstanceFleetByName | Construct | A Step Functions Task to modify an InstanceFleet on an EMR Cluster. |
EmrModifyInstanceFleetByNameJsonataProps | Struct | Properties for EmrModifyInstanceFleetByName using JSONata. |
EmrModifyInstanceFleetByNameJsonPathProps | Struct | Properties for EmrModifyInstanceFleetByName using JSONPath. |
EmrModifyInstanceFleetByNameProps | Struct | Properties for EmrModifyInstanceFleetByName. |
EmrModifyInstanceGroupByName | Construct | A Step Functions Task to modify an InstanceGroup on an EMR Cluster. |
EmrModifyInstanceGroupByName.InstanceGroupModifyConfigProperty | Struct | Modify the size or configurations of an instance group. |
EmrModifyInstanceGroupByName.InstanceResizePolicyProperty | Struct | Custom policy for requesting termination protection or termination of specific instances when shrinking an instance group. |
EmrModifyInstanceGroupByName.ShrinkPolicyProperty | Struct | Policy for customizing shrink operations. |
EmrModifyInstanceGroupByNameJsonataProps | Struct | Properties for EmrModifyInstanceGroupByName using JSONata. |
EmrModifyInstanceGroupByNameJsonPathProps | Struct | Properties for EmrModifyInstanceGroupByName using JSONPath. |
EmrModifyInstanceGroupByNameProps | Struct | Properties for EmrModifyInstanceGroupByName. |
EmrSetClusterTerminationProtection | Construct | A Step Functions Task to set Termination Protection on an EMR Cluster. |
EmrSetClusterTerminationProtectionJsonataProps | Struct | Properties for EmrSetClusterTerminationProtection using JSONata. |
EmrSetClusterTerminationProtectionJsonPathProps | Struct | Properties for EmrSetClusterTerminationProtection using JSONPath. |
EmrSetClusterTerminationProtectionProps | Struct | Properties for EmrSetClusterTerminationProtection. |
EmrTerminateCluster | Construct | A Step Functions Task to terminate an EMR Cluster. |
EmrTerminateClusterJsonataProps | Struct | Properties for EmrTerminateCluster using JSONata. |
EmrTerminateClusterJsonPathProps | Struct | Properties for EmrTerminateCluster using JSONPath. |
EmrTerminateClusterProps | Struct | Properties for EmrTerminateCluster. |
EncryptionConfiguration | Struct | Encryption Configuration of the S3 bucket. |
EncryptionOption | Enum | Encryption Options of the S3 bucket. |
EvaluateExpression | Construct | A Step Functions Task to evaluate an expression. |
EvaluateExpressionProps | Struct | Properties for EvaluateExpression. |
EventBridgePutEvents | Construct | A StepFunctions Task to send events to an EventBridge event bus. |
EventBridgePutEventsEntry | Struct | An entry to be sent to EventBridge. |
EventBridgePutEventsJsonataProps | Struct | Properties for sending events with PutEvents using JSONata. |
EventBridgePutEventsJsonPathProps | Struct | Properties for sending events with PutEvents using JSONPath. |
EventBridgePutEventsProps | Struct | Properties for sending events with PutEvents. |
EventBridgeSchedulerCreateScheduleTask | Construct | Create a new AWS EventBridge Scheduler schedule. |
EventBridgeSchedulerCreateScheduleTaskJsonataProps | Struct | Properties for creating an AWS EventBridge Scheduler schedule using JSONata. |
EventBridgeSchedulerCreateScheduleTaskJsonPathProps | Struct | Properties for creating an AWS EventBridge Scheduler schedule using JSONPath. |
EventBridgeSchedulerCreateScheduleTaskProps | Struct | Properties for creating an AWS EventBridge Scheduler schedule. |
EventBridgeSchedulerTarget | Class | The target that EventBridge Scheduler will invoke. |
EventBridgeSchedulerTargetProps | Struct | Properties for EventBridgeSchedulerTarget. |
ExecutionClass | Enum | The excecution class of the job. |
GlueDataBrewStartJobRun | Construct | Start a Job run as a Task. |
GlueDataBrewStartJobRunJsonataProps | Struct | Properties for starting a job run with StartJobRun using JSONata. |
GlueDataBrewStartJobRunJsonPathProps | Struct | Properties for starting a job run with StartJobRun using JSONPath. |
GlueDataBrewStartJobRunProps | Struct | Properties for starting a job run with StartJobRun. |
GlueStartCrawlerRun | Construct | Starts an AWS Glue Crawler in a Task state. |
GlueStartCrawlerRunJsonataProps | Struct | Properties for starting an AWS Glue Crawler as a task that using JSONata. |
GlueStartCrawlerRunJsonPathProps | Struct | Properties for starting an AWS Glue Crawler as a task that using JSONPath. |
GlueStartCrawlerRunProps | Struct | Properties for starting an AWS Glue Crawler as a task. |
GlueStartJobRun | Construct | Starts an AWS Glue job in a Task state. |
GlueStartJobRunJsonataProps | Struct | Properties for starting an AWS Glue job as a task. |
GlueStartJobRunJsonPathProps | Struct | Properties for starting an AWS Glue job as a task. |
GlueStartJobRunProps | Struct | Properties for starting an AWS Glue job as a task. |
Guardrail | Class | Guradrail settings for BedrockInvokeModel. |
HttpInvoke | Construct | A Step Functions Task to call a public third-party API. |
HttpInvokeJsonataProps | Struct | Properties for calling an external HTTP endpoint with HttpInvoke using JSONata. |
HttpInvokeJsonPathProps | Struct | Properties for calling an external HTTP endpoint with HttpInvoke using JSONPath. |
HttpInvokeProps | Struct | Properties for calling an external HTTP endpoint with HttpInvoke. |
HttpMethod | Enum | Http Methods that API Gateway supports. |
HttpMethods | Enum | Method type of a EKS call. |
IBedrockCreateModelCustomizationJobVpcConfig | Interface | VPC configuration. |
IContainerDefinition | Interface | Configuration of the container used to host the model. |
IEcsLaunchTarget | Interface | An Amazon ECS launch type determines the type of infrastructure on which your tasks and services are hosted. |
InputMode | Enum | Input mode that the algorithm supports. |
ISageMakerTask | Interface | Task to train a machine learning model using Amazon SageMaker. |
JobDependency | Struct | An object representing an AWS Batch job dependency. |
JobDriver | Struct | Specify the driver that the EMR Containers job runs on. |
LambdaInvocationType | Enum | Invocation type of a Lambda. |
LambdaInvoke | Construct | Invoke a Lambda function as a Task. |
LambdaInvokeJsonataProps | Struct | Properties for invoking a Lambda function with LambdaInvoke using Jsonata. |
LambdaInvokeJsonPathProps | Struct | Properties for invoking a Lambda function with LambdaInvoke using JsonPath. |
LambdaInvokeProps | Struct | Properties for invoking a Lambda function with LambdaInvoke. |
LaunchTargetBindOptions | Struct | Options for binding a launch target to an ECS run job task. |
MediaConvertCreateJob | Construct | A Step Functions Task to create a job in MediaConvert. |
MediaConvertCreateJobJsonataProps | Struct | Properties for creating a MediaConvert Job using JSONata. |
MediaConvertCreateJobJsonPathProps | Struct | Properties for creating a MediaConvert Job using JSONPath. |
MediaConvertCreateJobProps | Struct | Properties for creating a MediaConvert Job. |
MessageAttribute | Struct | A message attribute to add to the SNS message. |
MessageAttributeDataType | Enum | The data type set for the SNS message attributes. |
MetricDefinition | Struct | Specifies the metric name and regular expressions used to parse algorithm logs. |
Mode | Enum | Specifies how many models the container hosts. |
ModelClientOptions | Struct | Configures the timeout and maximum number of retries for processing a transform job invocation. |
Monitoring | Struct | Configuration setting for monitoring. |
OutputBucketConfiguration | Struct | S3 bucket configuration for the output data. |
OutputDataConfig | Struct | Configures the S3 bucket where SageMaker will save the result of model training. |
ProductionVariant | Struct | Identifies a model that you want to host and the resources to deploy for hosting it. |
QueryExecutionContext | Struct | Database and data catalog context in which the query execution occurs. |
RecordWrapperType | Enum | Define the format of the input data. |
ReleaseLabel | Class | The Amazon EMR release version to use for the job run. |
ResourceConfig | Struct | Specifies the resources, ML compute instances, and ML storage volumes to deploy for model training. |
ResultConfiguration | Struct | Location of query result along with S3 bucket configuration. |
RetryPolicy | Struct | The information about the retry policy settings. |
S3DataDistributionType | Enum | S3 Data Distribution Type. |
S3DataSource | Struct | S3 location of the channel data. |
S3DataType | Enum | S3 Data Type. |
S3Location | Class | Constructs IS3Location objects. |
S3LocationBindOptions | Struct | Options for binding an S3 Location. |
S3LocationConfig | Struct | Stores information about the location of an object in Amazon S3. |
SageMakerCreateEndpoint | Construct | A Step Functions Task to create a SageMaker endpoint. |
SageMakerCreateEndpointConfig | Construct | A Step Functions Task to create a SageMaker endpoint configuration. |
SageMakerCreateEndpointConfigJsonataProps | Struct | Properties for creating an Amazon SageMaker endpoint configuration using JSONata. |
SageMakerCreateEndpointConfigJsonPathProps | Struct | Properties for creating an Amazon SageMaker endpoint configuration using JSONPath. |
SageMakerCreateEndpointConfigProps | Struct | Properties for creating an Amazon SageMaker endpoint configuration. |
SageMakerCreateEndpointJsonataProps | Struct | Properties for creating an Amazon SageMaker endpoint using JSONata. |
SageMakerCreateEndpointJsonPathProps | Struct | Properties for creating an Amazon SageMaker endpoint using JSONPath. |
SageMakerCreateEndpointProps | Struct | Properties for creating an Amazon SageMaker endpoint. |
SageMakerCreateModel | Construct | A Step Functions Task to create a SageMaker model. |
SageMakerCreateModelJsonataProps | Struct | Properties for creating an Amazon SageMaker model using JSONata. |
SageMakerCreateModelJsonPathProps | Struct | Properties for creating an Amazon SageMaker model using JSONPath. |
SageMakerCreateModelProps | Struct | Properties for creating an Amazon SageMaker model. |
SageMakerCreateTrainingJob | Construct | Class representing the SageMaker Create Training Job task. |
SageMakerCreateTrainingJobJsonataProps | Struct | Properties for creating an Amazon SageMaker training job using JSONata. |
SageMakerCreateTrainingJobJsonPathProps | Struct | Properties for creating an Amazon SageMaker training job using JSONPath. |
SageMakerCreateTrainingJobProps | Struct | Properties for creating an Amazon SageMaker training job. |
SageMakerCreateTransformJob | Construct | Class representing the SageMaker Create Transform Job task. |
SageMakerCreateTransformJobJsonataProps | Struct | Properties for creating an Amazon SageMaker transform job task using JSONata. |
SageMakerCreateTransformJobJsonPathProps | Struct | Properties for creating an Amazon SageMaker transform job task using JSONPath. |
SageMakerCreateTransformJobProps | Struct | Properties for creating an Amazon SageMaker transform job task. |
SageMakerUpdateEndpoint | Construct | A Step Functions Task to update a SageMaker endpoint. |
SageMakerUpdateEndpointJsonataProps | Struct | Properties for updating Amazon SageMaker endpoint using JSONata. |
SageMakerUpdateEndpointJsonPathProps | Struct | Properties for updating Amazon SageMaker endpoint using JSONPath. |
SageMakerUpdateEndpointProps | Struct | Properties for updating Amazon SageMaker endpoint. |
Schedule | Class | Schedule for EventBridge Scheduler. |
ShuffleConfig | Struct | Configuration for a shuffle option for input data in a channel. |
SnsPublish | Construct | A Step Functions Task to publish messages to SNS topic. |
SnsPublishJsonataProps | Struct | Properties for publishing a message to an SNS topic using JSONata. |
SnsPublishJsonPathProps | Struct | Properties for publishing a message to an SNS topic using JSONPath. |
SnsPublishProps | Struct | Properties for publishing a message to an SNS topic. |
SparkSubmitJobDriver | Struct | The information about job driver for Spark submit. |
SplitType | Enum | Method to use to split the transform job's data files into smaller batches. |
SqsSendMessage | Construct | A StepFunctions Task to send messages to SQS queue. |
SqsSendMessageJsonataProps | Struct | Properties for sending a message to an SQS queue using JSONata. |
SqsSendMessageJsonPathProps | Struct | Properties for sending a message to an SQS queue using JSONPath. |
SqsSendMessageProps | Struct | Properties for sending a message to an SQS queue. |
StepFunctionsInvokeActivity | Construct | A Step Functions Task to invoke an Activity worker. |
StepFunctionsInvokeActivityJsonataProps | Struct | Properties for invoking an Activity worker using JSONata. |
StepFunctionsInvokeActivityJsonPathProps | Struct | Properties for invoking an Activity worker using JSONPath. |
StepFunctionsInvokeActivityProps | Struct | Properties for invoking an Activity worker. |
StepFunctionsStartExecution | Construct | A Step Functions Task to call StartExecution on another state machine. |
StepFunctionsStartExecutionJsonataProps | Struct | Properties for StartExecution using JSONata. |
StepFunctionsStartExecutionJsonPathProps | Struct | Properties for StartExecution using JSONPath. |
StepFunctionsStartExecutionProps | Struct | Properties for StartExecution. |
StoppingCondition | Struct | Specifies a limit to how long a model training job can run. |
TaskEnvironmentVariable | Struct | An environment variable to be set in the container run as a task. |
TrainingBucketConfiguration | Struct | S3 bucket configuration for the training data. |
TransformDataSource | Struct | S3 location of the input data that the model can consume. |
TransformInput | Struct | Dataset to be transformed and the Amazon S3 location where it is stored. |
TransformOutput | Struct | S3 location where you want Amazon SageMaker to save the results from the transform job. |
TransformResources | Struct | ML compute instances for the transform job. |
TransformS3DataSource | Struct | Location of the channel data. |
URLEncodingFormat | Enum | The style used when applying URL encoding to array values. |
ValidationBucketConfiguration | Struct | S3 bucket configuration for the validation data. |
VirtualClusterInput | Class | Class that returns a virtual cluster's id depending on input type. |
VpcConfig | Struct | Specifies the VPC that you want your Amazon SageMaker training job to connect to. |
WorkerConfigurationProperty | Struct | Properties for the worker configuration. |
WorkerType | Enum | The type of predefined worker that is allocated when a job runs. |
WorkerTypeV2 | Class | The type of predefined worker that is allocated when a job runs. |