Parallel
Define a Parallel state in the state machine.
A Parallel state can be used to run one or more state machines at the same time.
The Result of a Parallel state is an array of the results of its substatemachines.
Example
ts
import { Stack } from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as sfn from 'aws-cdk-lib/aws-stepfunctions';
interface MyJobProps {
jobFlavor: string;
}
class MyJob extends sfn.StateMachineFragment {
public readonly startState: sfn.State;
public readonly endStates: sfn.INextable[];
constructor(parent: Construct, id: string, props: MyJobProps) {
super(parent, id);
const choice = new sfn.Choice(this, 'Choice')
.when(sfn.Condition.stringEquals('$.branch', 'left'), new sfn.Pass(this, 'Left Branch'))
.when(sfn.Condition.stringEquals('$.branch', 'right'), new sfn.Pass(this, 'Right Branch'));
// ...
this.startState = choice;
this.endStates = choice.afterwards().endStates;
}
}
class MyStack extends Stack {
constructor(scope: Construct, id: string) {
super(scope, id);
// Do 3 different variants of MyJob in parallel
const parallel = new sfn.Parallel(this, 'All jobs')
.branch(new MyJob(this, 'Quick', { jobFlavor: 'quick' }).prefixStates())
.branch(new MyJob(this, 'Medium', { jobFlavor: 'medium' }).prefixStates())
.branch(new MyJob(this, 'Slow', { jobFlavor: 'slow' }).prefixStates());
new sfn.StateMachine(this, 'MyStateMachine', {
definitionBody: sfn.DefinitionBody.fromChainable(parallel),
});
}
}
python
from aws_cdk import Stack
from constructs import Construct
import aws_cdk.aws_stepfunctions as sfn
class MyJob(sfn.StateMachineFragment):
def __init__(self, parent, id, *, jobFlavor):
super().__init__(parent, id)
choice = sfn.Choice(self, "Choice").when(sfn.Condition.string_equals("$.branch", "left"), sfn.Pass(self, "Left Branch")).when(sfn.Condition.string_equals("$.branch", "right"), sfn.Pass(self, "Right Branch"))
# ...
self.start_state = choice
self.end_states = choice.afterwards().end_states
class MyStack(Stack):
def __init__(self, scope, id):
super().__init__(scope, id)
# Do 3 different variants of MyJob in parallel
parallel = sfn.Parallel(self, "All jobs").branch(MyJob(self, "Quick", job_flavor="quick").prefix_states()).branch(MyJob(self, "Medium", job_flavor="medium").prefix_states()).branch(MyJob(self, "Slow", job_flavor="slow").prefix_states())
sfn.StateMachine(self, "MyStateMachine",
definition_body=sfn.DefinitionBody.from_chainable(parallel)
)
java
import software.amazon.awscdk.Stack;
import software.constructs.Construct;
import software.amazon.awscdk.services.stepfunctions.*;
public class MyJobProps {
private String jobFlavor;
public String getJobFlavor() {
return this.jobFlavor;
}
public MyJobProps jobFlavor(String jobFlavor) {
this.jobFlavor = jobFlavor;
return this;
}
}
public class MyJob extends StateMachineFragment {
public final State startState;
public final INextable[] endStates;
public MyJob(Construct parent, String id, MyJobProps props) {
super(parent, id);
Choice choice = new Choice(this, "Choice").when(Condition.stringEquals("$.branch", "left"), new Pass(this, "Left Branch")).when(Condition.stringEquals("$.branch", "right"), new Pass(this, "Right Branch"));
// ...
this.startState = choice;
this.endStates = choice.afterwards().getEndStates();
}
}
public class MyStack extends Stack {
public MyStack(Construct scope, String id) {
super(scope, id);
// Do 3 different variants of MyJob in parallel
Parallel parallel = new Parallel(this, "All jobs").branch(new MyJob(this, "Quick", new MyJobProps().jobFlavor("quick")).prefixStates()).branch(new MyJob(this, "Medium", new MyJobProps().jobFlavor("medium")).prefixStates()).branch(new MyJob(this, "Slow", new MyJobProps().jobFlavor("slow")).prefixStates());
StateMachine.Builder.create(this, "MyStateMachine")
.definitionBody(DefinitionBody.fromChainable(parallel))
.build();
}
}
csharp
// Install: dotnet add package Amazon.CDK.Lib
using Amazon.CDK;
using Constructs;
using Amazon.CDK.AWS.StepFunctions;
class MyJobProps
{
public string JobFlavor { get; set; }
}
class MyJob : StateMachineFragment
{
public State StartState { get; }
public INextable[] EndStates { get; }
public MyJob(Construct parent, string id, MyJobProps props) : base(parent, id)
{
var choice = new Choice(this, "Choice").When(Condition.StringEquals("$.branch", "left"), new Pass(this, "Left Branch")).When(Condition.StringEquals("$.branch", "right"), new Pass(this, "Right Branch"));
// ...
StartState = choice;
EndStates = choice.Afterwards().EndStates;
}
}
class MyStack : Stack
{
public MyStack(Construct scope, string id) : base(scope, id)
{
// Do 3 different variants of MyJob in parallel
var parallel = new Parallel(this, "All jobs").Branch(new MyJob(this, "Quick", new MyJobProps { JobFlavor = "quick" }).PrefixStates()).Branch(new MyJob(this, "Medium", new MyJobProps { JobFlavor = "medium" }).PrefixStates()).Branch(new MyJob(this, "Slow", new MyJobProps { JobFlavor = "slow" }).PrefixStates());
new StateMachine(this, "MyStateMachine", new StateMachineProps {
DefinitionBody = DefinitionBody.FromChainable(parallel)
});
}
}
go
// Install: go get github.com/aws/aws-cdk-go/awscdk/v2
import "github.com/aws/aws-cdk-go/awscdk"
import "github.com/aws/constructs-go/constructs"
import "github.com/aws/aws-cdk-go/awscdk"
type myJobProps struct {
jobFlavor *string
}
type myJob struct {
StateMachineFragment
startState State
endStates []INextable
}
func newMyJob(parent Construct, id *string, props myJobProps) *myJob {
this := &myJob{}
sfn.NewStateMachineFragment_Override(this, parent, id)
choice := sfn.NewChoice(this, jsii.String("Choice")).When(sfn.Condition_StringEquals(jsii.String("$.branch"), jsii.String("left")), sfn.NewPass(this, jsii.String("Left Branch"))).When(sfn.Condition_StringEquals(jsii.String("$.branch"), jsii.String("right")), sfn.NewPass(this, jsii.String("Right Branch")))
// ...
this.startState = choice
this.endStates = choice.Afterwards().EndStates
return this
}
type myStack struct {
Stack
}
func newMyStack(scope Construct, id *string) *myStack {
this := &myStack{}
newStack_Override(this, scope, id)
// Do 3 different variants of MyJob in parallel
parallel := sfn.NewParallel(this, jsii.String("All jobs")).Branch(NewMyJob(this, jsii.String("Quick"), &myJobProps{
jobFlavor: jsii.String("quick"),
}).PrefixStates()).Branch(NewMyJob(this, jsii.String("Medium"), &myJobProps{
jobFlavor: jsii.String("medium"),
}).PrefixStates()).Branch(NewMyJob(this, jsii.String("Slow"), &myJobProps{
jobFlavor: jsii.String("slow"),
}).PrefixStates())
sfn.NewStateMachine(this, jsii.String("MyStateMachine"), &StateMachineProps{
DefinitionBody: sfn.DefinitionBody_FromChainable(parallel),
})
return this
}
Initializer
ts
new Parallel(scope: Construct, id: string, props?: ParallelProps)
python
Parallel(scope: Construct, id: str, props: typing.Optional[ParallelProps] = None)
java
new Parallel(Construct scope, String id, ParallelProps props)
csharp
new Parallel(Construct scope, string id, ParallelProps? props = null)
go
NewParallel(Scope Construct, Id *string, Props ParallelProps)
Construct Props
arguments{ [key: string]: any }argumentstyping.Mapping[str, Any]argumentsMap<String, Object>ArgumentsIDictionary<string, object>Arguments*map[string]interface{}#Parameters pass a collection of key-value pairs, either static values or JSONata expressions that select from the input.
Default: No arguments
assign{ [key: string]: any }assigntyping.Mapping[str, Any]assignMap<String, Object>AssignIDictionary<string, object>Assign*map[string]interface{}#Workflow variables to store in this step.
Default: Not assign variables
Using workflow variables, you can store data in a step and retrieve that data in future steps.
A comment describing this state.
Default: No comment
JSONPath expression to select part of the state to be the input to this state.
May also be the special value JsonPath.DISCARD, which will cause the effective input to be the empty object {}.
JSONPath expression to select part of the state to be the output to this state.
May also be the special value JsonPath.DISCARD, which will cause the effective output to be the empty object {}.
Used to specify and transform output from the state.
Default: $states.result or $states.errorOutput
notes
When specified, the value overrides the state output default. The output field accepts any JSON value (object, array, string, number, boolean, null). Any string value, including those inside objects or arrays, will be evaluated as JSONata if surrounded by {% %} characters. Output also accepts a JSONata expression directly.
parameters{ [key: string]: any }parameterstyping.Mapping[str, Any]parametersMap<String, Object>ParametersIDictionary<string, object>Parameters*map[string]interface{}#Parameters pass a collection of key-value pairs, either static values or JSONPath expressions that select from the input.
Default: No parameters
queryLanguageQueryLanguagequery_languageQueryLanguagequeryLanguageQueryLanguageQueryLanguageQueryLanguageQueryLanguageQueryLanguage= JSONPath#The name of the query language used by the state.
If the state does not contain a queryLanguage field, then it will use the query language specified in the top-level queryLanguage field.
JSONPath expression to indicate where to inject the state's output.
May also be the special value JsonPath.DISCARD, which will cause the state's input to become its output.
resultSelector{ [key: string]: any }result_selectortyping.Mapping[str, Any]resultSelectorMap<String, Object>ResultSelectorIDictionary<string, object>ResultSelector*map[string]interface{}= None#The JSON that will replace the state's raw result and become the effective result before ResultPath is applied.
You can use ResultSelector to create a payload with values that are static or selected from the state's raw result.
Optional name for this state.
Default: The construct ID will be used as state name
Properties
Methods
addCatch(handler: IChainable, props?: CatchProps): Parallel // chainabledef add_catch(self, handler: IChainable, props: typing.Optional[CatchProps] = None) -> Parallelpublic Parallel addCatch(IChainable handler, CatchProps props)public Parallel AddCatch(IChainable handler, CatchProps? props = null)func AddCatch(Handler IChainable, Props CatchProps) Parallel#Add a recovery handler for this state.
When a particular error occurs, execution will continue at the error handler instead of failing the state machine execution.
addPrefix(x: string): voiddef add_prefix(self, x: str) -> Nonepublic void addPrefix(String x)public void AddPrefix(string x)func AddPrefix(X *string)#Add a prefix to the stateId of this state.
addRetry(props?: RetryProps): Parallel // chainabledef add_retry(self, props: typing.Optional[RetryProps] = None) -> Parallelpublic Parallel addRetry(RetryProps props)public Parallel AddRetry(RetryProps? props = null)func AddRetry(Props RetryProps) Parallel#Add retry configuration for this state.
This controls if and how the execution will be retried if a particular error occurs.
bindToGraph(graph: StateGraph): voiddef bind_to_graph(self, graph: StateGraph) -> Nonepublic void bindToGraph(StateGraph graph)public void BindToGraph(StateGraph graph)func BindToGraph(Graph StateGraph)#Overwrites State.bindToGraph. Adds branches to the Parallel state here so that any necessary prefixes are appended first.
branch(...branches: IChainable[]): Parallel // chainabledef branch(self, branches: IChainable) -> Parallelpublic Parallel branch(IChainable branches)public Parallel Branch(IChainable branches)func Branch(Branches IChainable) Parallel#Define one or more branches to run in parallel.
filterNextables(states: State[]): INextable[]@staticmethod def filter_nextables(states: typing.List[State]) -> typing.List[INextable]public static List<INextable> filterNextables(List<State> states)public static INextable[] FilterNextables(State[] states)func FilterNextables(States *[]State) *[]INextablestatic#Return only the states that allow chaining from an array of states.
findReachableEndStates(start: State, options?: FindStateOptions): State[]@staticmethod def find_reachable_end_states(start: State, options: typing.Optional[FindStateOptions] = None) -> typing.List[State]public static List<State> findReachableEndStates(State start, FindStateOptions options)public static State[] FindReachableEndStates(State start, FindStateOptions? options = null)func FindReachableEndStates(Start State, Options FindStateOptions) *[]Statestatic#Find the set of end states states reachable through transitions from the given start state.
findReachableStates(start: State, options?: FindStateOptions): State[]@staticmethod def find_reachable_states(start: State, options: typing.Optional[FindStateOptions] = None) -> typing.List[State]public static List<State> findReachableStates(State start, FindStateOptions options)public static State[] FindReachableStates(State start, FindStateOptions? options = null)func FindReachableStates(Start State, Options FindStateOptions) *[]Statestatic#Find the set of states reachable through transitions from the given start state.
This does not retrieve states from within sub-graphs, such as states within a Parallel state's branch.
isConstruct(x: any): boolean@staticmethod def is_construct(x: Any) -> boolpublic static Boolean isConstruct(Object x)public static bool IsConstruct(object x)func IsConstruct(X interface{}) *boolstatic#Checks if `x` is a construct.
notes · 1 parameter · returns
Use this method instead of instanceof to properly detect Construct instances, even when the construct library is symlinked.
Explanation: in JavaScript, multiple copies of the constructs library on disk are seen as independent, completely different libraries. As a consequence, the class Construct in each copy of the constructs library is seen as a different class, and an instance of one class will not test as instanceof the other class. npm install will not create installations like this, but users may manually symlink construct libraries together or use a monorepo tool: in those cases, multiple copies of the constructs library can be accidentally installed, and instanceof will behave unpredictably. It is safest to avoid using instanceof, and using this type-testing method instead.
Parameters
xxxxX— Any object.
Returns: true if x is an object created from a class which extends Construct.
jsonata(scope: Construct, id: string, props?: ParallelJsonataProps): Parallel // chainable@staticmethod def jsonata(scope: Construct, id: str, props: typing.Optional[ParallelJsonataProps] = None) -> Parallelpublic static Parallel jsonata(Construct scope, String id, ParallelJsonataProps props)public static Parallel Jsonata(Construct scope, string id, ParallelJsonataProps? props = null)func Jsonata(Scope Construct, Id *string, Props ParallelJsonataProps) Parallelstatic#Define a Parallel state using JSONata in the state machine.
A Parallel state can be used to run one or more state machines at the same time.
The Result of a Parallel state is an array of the results of its substatemachines.
jsonPath(scope: Construct, id: string, props?: ParallelJsonPathProps): Parallel // chainable@staticmethod def json_path(scope: Construct, id: str, props: typing.Optional[ParallelJsonPathProps] = None) -> Parallelpublic static Parallel jsonPath(Construct scope, String id, ParallelJsonPathProps props)public static Parallel JsonPath(Construct scope, string id, ParallelJsonPathProps? props = null)func JsonPath(Scope Construct, Id *string, Props ParallelJsonPathProps) Parallelstatic#Define a Parallel state using JSONPath in the state machine.
A Parallel state can be used to run one or more state machines at the same time.
The Result of a Parallel state is an array of the results of its substatemachines.
next(next: IChainable): Chaindef next(self, next: IChainable) -> Chainpublic Chain next(IChainable next)public Chain Next(IChainable next)func Next(Next IChainable) Chain#Continue normal execution with the given state.
prefixStates(root: IConstruct, prefix: string): void@staticmethod def prefix_states(root: IConstruct, prefix: str) -> Nonepublic static void prefixStates(IConstruct root, String prefix)public static void PrefixStates(IConstruct root, string prefix)func PrefixStates(Root IConstruct, Prefix *string)static#Add a prefix to the stateId of all States found in a construct tree.
toStateJson(topLevelQueryLanguage?: QueryLanguage): jsondef to_state_json(self, top_level_query_language: typing.Optional[QueryLanguage] = None) -> Anypublic Object toStateJson(QueryLanguage topLevelQueryLanguage)public object ToStateJson(QueryLanguage? topLevelQueryLanguage = null)func ToStateJson(TopLevelQueryLanguage QueryLanguage) interface{}#Return the Amazon States Language object for this state.
toString(): stringdef to_string(self, ) -> strpublic String toString()public string ToString()func ToString() *string#Returns a string representation of this construct.
with(...mixins: IMixin[]): IConstructdef with(self, mixins: IMixin) -> IConstructpublic IConstruct with(IMixin mixins)public IConstruct With(IMixin mixins)func With(Mixins IMixin) IConstruct#Applies one or more mixins to this construct.
Mixins are applied in order. The list of constructs is captured at the start of the call, so constructs added by a mixin will not be visited. Use multiple with() calls if subsequent mixins should apply to added constructs.
1 parameter · returns
Parameters
mixinsmixinsmixinsmixinsMixins— The mixins to apply.
Returns: This construct for chaining