Repository: awslabs/aws-ddk (language: python)

Issue:
core: EMR Serverless Stage
create s step function stage that can trigger an EMR serverless job if the job is not provided, create an EMR serverless job.
core: EMR Serverless Cluster Environment
Create a core construct for an EMR serverless cluster 


Discussion / hints:

Fix patch:
diff --git a/API.md b/API.md
index 585c3537..cc68208c 100644
--- a/API.md
+++ b/API.md
@@ -3533,6 +3533,234 @@ public readonly vpc: IVpc;
 ---
 
 
+### EMRServerlessJobStage <a name="EMRServerlessJobStage" id="aws-ddk-core.EMRServerlessJobStage"></a>
+
+Stage that contains a step function that runs an EMR Job.
+
+#### Initializers <a name="Initializers" id="aws-ddk-core.EMRServerlessJobStage.Initializer"></a>
+
+```typescript
+import { EMRServerlessJobStage } from 'aws-ddk-core'
+
+new EMRServerlessJobStage(scope: Construct, id: string, props: EMRServerlessJobStageProps)
+```
+
+| **Name** | **Type** | **Description** |
+| --- | --- | --- |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.Initializer.parameter.scope">scope</a></code> | <code>constructs.Construct</code> | Scope within which this construct is defined. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.Initializer.parameter.id">id</a></code> | <code>string</code> | Identifier of the stage. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.Initializer.parameter.props">props</a></code> | <code><a href="#aws-ddk-core.EMRServerlessJobStageProps">EMRServerlessJobStageProps</a></code> | Properties for the stage. |
+
+---
+
+##### `scope`<sup>Required</sup> <a name="scope" id="aws-ddk-core.EMRServerlessJobStage.Initializer.parameter.scope"></a>
+
+- *Type:* constructs.Construct
+
+Scope within which this construct is defined.
+
+---
+
+##### `id`<sup>Required</sup> <a name="id" id="aws-ddk-core.EMRServerlessJobStage.Initializer.parameter.id"></a>
+
+- *Type:* string
+
+Identifier of the stage.
+
+---
+
+##### `props`<sup>Required</sup> <a name="props" id="aws-ddk-core.EMRServerlessJobStage.Initializer.parameter.props"></a>
+
+- *Type:* <a href="#aws-ddk-core.EMRServerlessJobStageProps">EMRServerlessJobStageProps</a>
+
+Properties for the stage.
+
+---
+
+#### Methods <a name="Methods" id="Methods"></a>
+
+| **Name** | **Description** |
+| --- | --- |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.toString">toString</a></code> | Returns a string representation of this construct. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.addAlarm">addAlarm</a></code> | Add a CloudWatch alarm for the DataStage. |
+
+---
+
+##### `toString` <a name="toString" id="aws-ddk-core.EMRServerlessJobStage.toString"></a>
+
+```typescript
+public toString(): string
+```
+
+Returns a string representation of this construct.
+
+##### `addAlarm` <a name="addAlarm" id="aws-ddk-core.EMRServerlessJobStage.addAlarm"></a>
+
+```typescript
+public addAlarm(id: string, props: AlarmProps): DataStage
+```
+
+Add a CloudWatch alarm for the DataStage.
+
+###### `id`<sup>Required</sup> <a name="id" id="aws-ddk-core.EMRServerlessJobStage.addAlarm.parameter.id"></a>
+
+- *Type:* string
+
+Identifier of the CloudWatch Alarm.
+
+---
+
+###### `props`<sup>Required</sup> <a name="props" id="aws-ddk-core.EMRServerlessJobStage.addAlarm.parameter.props"></a>
+
+- *Type:* <a href="#aws-ddk-core.AlarmProps">AlarmProps</a>
+
+Properties for the alarm.
+
+---
+
+#### Static Functions <a name="Static Functions" id="Static Functions"></a>
+
+| **Name** | **Description** |
+| --- | --- |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.isConstruct">isConstruct</a></code> | Checks if `x` is a construct. |
+
+---
+
+##### ~~`isConstruct`~~ <a name="isConstruct" id="aws-ddk-core.EMRServerlessJobStage.isConstruct"></a>
+
+```typescript
+import { EMRServerlessJobStage } from 'aws-ddk-core'
+
+EMRServerlessJobStage.isConstruct(x: any)
+```
+
+Checks if `x` is a construct.
+
+###### `x`<sup>Required</sup> <a name="x" id="aws-ddk-core.EMRServerlessJobStage.isConstruct.parameter.x"></a>
+
+- *Type:* any
+
+Any object.
+
+---
+
+#### Properties <a name="Properties" id="Properties"></a>
+
+| **Name** | **Type** | **Description** |
+| --- | --- | --- |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.property.node">node</a></code> | <code>constructs.Node</code> | The tree node. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.property.description">description</a></code> | <code>string</code> | Description of the stage. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.property.eventPattern">eventPattern</a></code> | <code>aws-cdk-lib.aws_events.EventPattern</code> | Output event pattern of the stage. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.property.name">name</a></code> | <code>string</code> | Name of the stage. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.property.targets">targets</a></code> | <code>aws-cdk-lib.aws_events.IRuleTarget[]</code> | Input targets for the stage. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.property.alarmsEnabled">alarmsEnabled</a></code> | <code>boolean</code> | Flag indicating whether the alarms are enabled for this stage. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.property.cloudwatchAlarms">cloudwatchAlarms</a></code> | <code>aws-cdk-lib.aws_cloudwatch.Alarm[]</code> | List of CloudWatch Alarms linked to the stage. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStage.property.stateMachine">stateMachine</a></code> | <code>aws-cdk-lib.aws_stepfunctions.StateMachine</code> | State machine. |
+
+---
+
+##### `node`<sup>Required</sup> <a name="node" id="aws-ddk-core.EMRServerlessJobStage.property.node"></a>
+
+```typescript
+public readonly node: Node;
+```
+
+- *Type:* constructs.Node
+
+The tree node.
+
+---
+
+##### `description`<sup>Optional</sup> <a name="description" id="aws-ddk-core.EMRServerlessJobStage.property.description"></a>
+
+```typescript
+public readonly description: string;
+```
+
+- *Type:* string
+
+Description of the stage.
+
+---
+
+##### `eventPattern`<sup>Optional</sup> <a name="eventPattern" id="aws-ddk-core.EMRServerlessJobStage.property.eventPattern"></a>
+
+```typescript
+public readonly eventPattern: EventPattern;
+```
+
+- *Type:* aws-cdk-lib.aws_events.EventPattern
+
+Output event pattern of the stage.
+
+Event pattern describes the structure of output event(s) produced by this stage.
+Event Rules use event patterns to select events and route them to targets.
+
+---
+
+##### `name`<sup>Optional</sup> <a name="name" id="aws-ddk-core.EMRServerlessJobStage.property.name"></a>
+
+```typescript
+public readonly name: string;
+```
+
+- *Type:* string
+
+Name of the stage.
+
+---
+
+##### `targets`<sup>Optional</sup> <a name="targets" id="aws-ddk-core.EMRServerlessJobStage.property.targets"></a>
+
+```typescript
+public readonly targets: IRuleTarget[];
+```
+
+- *Type:* aws-cdk-lib.aws_events.IRuleTarget[]
+
+Input targets for the stage.
+
+Targets are used by Event Rules to describe what should be invoked when a rule matches an event.
+
+---
+
+##### `alarmsEnabled`<sup>Required</sup> <a name="alarmsEnabled" id="aws-ddk-core.EMRServerlessJobStage.property.alarmsEnabled"></a>
+
+```typescript
+public readonly alarmsEnabled: boolean;
+```
+
+- *Type:* boolean
+
+Flag indicating whether the alarms are enabled for this stage.
+
+---
+
+##### `cloudwatchAlarms`<sup>Required</sup> <a name="cloudwatchAlarms" id="aws-ddk-core.EMRServerlessJobStage.property.cloudwatchAlarms"></a>
+
+```typescript
+public readonly cloudwatchAlarms: Alarm[];
+```
+
+- *Type:* aws-cdk-lib.aws_cloudwatch.Alarm[]
+
+List of CloudWatch Alarms linked to the stage.
+
+---
+
+##### `stateMachine`<sup>Required</sup> <a name="stateMachine" id="aws-ddk-core.EMRServerlessJobStage.property.stateMachine"></a>
+
+```typescript
+public readonly stateMachine: StateMachine;
+```
+
+- *Type:* aws-cdk-lib.aws_stepfunctions.StateMachine
+
+State machine.
+
+---
+
+
 ### EventStage <a name="EventStage" id="aws-ddk-core.EventStage"></a>
 
 Class that represents an event stage within a data pipeline.
@@ -9115,6 +9343,233 @@ Existing vpc id.
 
 ---
 
+### EMRServerlessJobStageProps <a name="EMRServerlessJobStageProps" id="aws-ddk-core.EMRServerlessJobStageProps"></a>
+
+Properties of the EMR Serverless Job stage.
+
+#### Initializer <a name="Initializer" id="aws-ddk-core.EMRServerlessJobStageProps.Initializer"></a>
+
+```typescript
+import { EMRServerlessJobStageProps } from 'aws-ddk-core'
+
+const eMRServerlessJobStageProps: EMRServerlessJobStageProps = { ... }
+```
+
+#### Properties <a name="Properties" id="Properties"></a>
+
+| **Name** | **Type** | **Description** |
+| --- | --- | --- |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.description">description</a></code> | <code>string</code> | Description of the stage. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.name">name</a></code> | <code>string</code> | Name of the stage. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.additionalRolePolicyStatements">additionalRolePolicyStatements</a></code> | <code>aws-cdk-lib.aws_iam.PolicyStatement[]</code> | Additional IAM policy statements to add to the state machine role. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.alarmsEnabled">alarmsEnabled</a></code> | <code>boolean</code> | Enable/Disable all alarms in the stage. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.definition">definition</a></code> | <code>string \| aws-cdk-lib.aws_stepfunctions.IChainable</code> | Steps for the state machine. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.definitionFile">definitionFile</a></code> | <code>string</code> | File containing a JSON definition for the state machine. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.stateMachineFailedExecutionsAlarmEvaluationPeriods">stateMachineFailedExecutionsAlarmEvaluationPeriods</a></code> | <code>number</code> | The number of periods over which data is compared to the specified threshold. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.stateMachineFailedExecutionsAlarmThreshold">stateMachineFailedExecutionsAlarmThreshold</a></code> | <code>number</code> | The number of failed state machine executions before triggering CW alarm. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.stateMachineInput">stateMachineInput</a></code> | <code>{[ key: string ]: any}</code> | Input of the state machine. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.stateMachineName">stateMachineName</a></code> | <code>string</code> | Name of the state machine. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.applicationId">applicationId</a></code> | <code>string</code> | EMR Serverless Application Id. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.executionRoleArn">executionRoleArn</a></code> | <code>string</code> | EMR Execution Role Arn. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.jobDriver">jobDriver</a></code> | <code>{[ key: string ]: any}</code> | The job driver for the job run. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.jobExecutionStatusCheckPeriod">jobExecutionStatusCheckPeriod</a></code> | <code>aws-cdk-lib.Duration</code> | Duration to wait between polling job status. |
+| <code><a href="#aws-ddk-core.EMRServerlessJobStageProps.property.startJobRunProps">startJobRunProps</a></code> | <code>{[ key: string ]: any}</code> | Additional properties to pass to 'emrserverless:StartJobRun'. |
+
+---
+
+##### `description`<sup>Optional</sup> <a name="description" id="aws-ddk-core.EMRServerlessJobStageProps.property.description"></a>
+
+```typescript
+public readonly description: string;
+```
+
+- *Type:* string
+
+Description of the stage.
+
+---
+
+##### `name`<sup>Optional</sup> <a name="name" id="aws-ddk-core.EMRServerlessJobStageProps.property.name"></a>
+
+```typescript
+public readonly name: string;
+```
+
+- *Type:* string
+
+Name of the stage.
+
+---
+
+##### `additionalRolePolicyStatements`<sup>Optional</sup> <a name="additionalRolePolicyStatements" id="aws-ddk-core.EMRServerlessJobStageProps.property.additionalRolePolicyStatements"></a>
+
+```typescript
+public readonly additionalRolePolicyStatements: PolicyStatement[];
+```
+
+- *Type:* aws-cdk-lib.aws_iam.PolicyStatement[]
+
+Additional IAM policy statements to add to the state machine role.
+
+---
+
+##### `alarmsEnabled`<sup>Optional</sup> <a name="alarmsEnabled" id="aws-ddk-core.EMRServerlessJobStageProps.property.alarmsEnabled"></a>
+
+```typescript
+public readonly alarmsEnabled: boolean;
+```
+
+- *Type:* boolean
+- *Default:* true
+
+Enable/Disable all alarms in the stage.
+
+---
+
+##### `definition`<sup>Optional</sup> <a name="definition" id="aws-ddk-core.EMRServerlessJobStageProps.property.definition"></a>
+
+```typescript
+public readonly definition: string | IChainable;
+```
+
+- *Type:* string | aws-cdk-lib.aws_stepfunctions.IChainable
+
+Steps for the state machine.
+
+Can either be provided as 'sfn.IChainable' or a JSON string.
+
+---
+
+##### `definitionFile`<sup>Optional</sup> <a name="definitionFile" id="aws-ddk-core.EMRServerlessJobStageProps.property.definitionFile"></a>
+
+```typescript
+public readonly definitionFile: string;
+```
+
+- *Type:* string
+
+File containing a JSON definition for the state machine.
+
+---
+
+##### `stateMachineFailedExecutionsAlarmEvaluationPeriods`<sup>Optional</sup> <a name="stateMachineFailedExecutionsAlarmEvaluationPeriods" id="aws-ddk-core.EMRServerlessJobStageProps.property.stateMachineFailedExecutionsAlarmEvaluationPeriods"></a>
+
+```typescript
+public readonly stateMachineFailedExecutionsAlarmEvaluationPeriods: number;
+```
+
+- *Type:* number
+- *Default:* 1
+
+The number of periods over which data is compared to the specified threshold.
+
+---
+
+##### `stateMachineFailedExecutionsAlarmThreshold`<sup>Optional</sup> <a name="stateMachineFailedExecutionsAlarmThreshold" id="aws-ddk-core.EMRServerlessJobStageProps.property.stateMachineFailedExecutionsAlarmThreshold"></a>
+
+```typescript
+public readonly stateMachineFailedExecutionsAlarmThreshold: number;
+```
+
+- *Type:* number
+- *Default:* 1
+
+The number of failed state machine executions before triggering CW alarm.
+
+---
+
+##### `stateMachineInput`<sup>Optional</sup> <a name="stateMachineInput" id="aws-ddk-core.EMRServerlessJobStageProps.property.stateMachineInput"></a>
+
+```typescript
+public readonly stateMachineInput: {[ key: string ]: any};
+```
+
+- *Type:* {[ key: string ]: any}
+
+Input of the state machine.
+
+---
+
+##### `stateMachineName`<sup>Optional</sup> <a name="stateMachineName" id="aws-ddk-core.EMRServerlessJobStageProps.property.stateMachineName"></a>
+
+```typescript
+public readonly stateMachineName: string;
+```
+
+- *Type:* string
+
+Name of the state machine.
+
+---
+
+##### `applicationId`<sup>Required</sup> <a name="applicationId" id="aws-ddk-core.EMRServerlessJobStageProps.property.applicationId"></a>
+
+```typescript
+public readonly applicationId: string;
+```
+
+- *Type:* string
+
+EMR Serverless Application Id.
+
+---
+
+##### `executionRoleArn`<sup>Required</sup> <a name="executionRoleArn" id="aws-ddk-core.EMRServerlessJobStageProps.property.executionRoleArn"></a>
+
+```typescript
+public readonly executionRoleArn: string;
+```
+
+- *Type:* string
+
+EMR Execution Role Arn.
+
+---
+
+##### `jobDriver`<sup>Required</sup> <a name="jobDriver" id="aws-ddk-core.EMRServerlessJobStageProps.property.jobDriver"></a>
+
+```typescript
+public readonly jobDriver: {[ key: string ]: any};
+```
+
+- *Type:* {[ key: string ]: any}
+
+The job driver for the job run.
+
+This is a Tagged Union structure.
+Only one of the following top level
+keys can be set: 'sparkSubmit', 'hive'
+
+---
+
+##### `jobExecutionStatusCheckPeriod`<sup>Optional</sup> <a name="jobExecutionStatusCheckPeriod" id="aws-ddk-core.EMRServerlessJobStageProps.property.jobExecutionStatusCheckPeriod"></a>
+
+```typescript
+public readonly jobExecutionStatusCheckPeriod: Duration;
+```
+
+- *Type:* aws-cdk-lib.Duration
+
+Duration to wait between polling job status.
+
+Defaults to 30 seconds.
+
+---
+
+##### `startJobRunProps`<sup>Optional</sup> <a name="startJobRunProps" id="aws-ddk-core.EMRServerlessJobStageProps.property.startJobRunProps"></a>
+
+```typescript
+public readonly startJobRunProps: {[ key: string ]: any};
+```
+
+- *Type:* {[ key: string ]: any}
+
+Additional properties to pass to 'emrserverless:StartJobRun'.
+
+https://docs.aws.amazon.com/emr-serverless/latest/APIReference/API_StartJobRun.html
+
+---
+
 ### EnvironmentConfiguration <a name="EnvironmentConfiguration" id="aws-ddk-core.EnvironmentConfiguration"></a>
 
 #### Initializer <a name="Initializer" id="aws-ddk-core.EnvironmentConfiguration.Initializer"></a>
diff --git a/src/stages/emr-serverless-job.ts b/src/stages/emr-serverless-job.ts
new file mode 100644
index 00000000..f0a7c59d
--- /dev/null
+++ b/src/stages/emr-serverless-job.ts
@@ -0,0 +1,123 @@
+import * as cdk from "aws-cdk-lib";
+import { Duration } from "aws-cdk-lib";
+import * as events from "aws-cdk-lib/aws-events";
+import * as iam from "aws-cdk-lib/aws-iam";
+import * as sfn from "aws-cdk-lib/aws-stepfunctions";
+import * as tasks from "aws-cdk-lib/aws-stepfunctions-tasks";
+import { Construct } from "constructs";
+import { StateMachineStage, StateMachineStageProps } from "../pipelines";
+/**
+ * Properties of the EMR Serverless Job stage.
+ */
+export interface EMRServerlessJobStageProps extends StateMachineStageProps {
+  /**
+   * EMR Serverless Application Id.
+   */
+  readonly applicationId: string;
+  /**
+   * EMR Execution Role Arn.
+   */
+  readonly executionRoleArn: string;
+  /**
+   * The job driver for the job run.
+   * This is a Tagged Union structure.
+   * Only one of the following top level
+   * keys can be set: 'sparkSubmit', 'hive'
+   */
+  readonly jobDriver: { [key: string]: any };
+  /**
+   * Duration to wait between polling job status.
+   * Defaults to 30 seconds.
+   */
+  readonly jobExecutionStatusCheckPeriod?: cdk.Duration;
+  /**
+   * Additional properties to pass to 'emrserverless:StartJobRun'.
+   * https://docs.aws.amazon.com/emr-serverless/latest/APIReference/API_StartJobRun.html
+   */
+  readonly startJobRunProps?: { [key: string]: any };
+}
+
+/**
+ * Stage that contains a step function that runs an EMR Job.
+ */
+export class EMRServerlessJobStage extends StateMachineStage {
+  readonly targets?: events.IRuleTarget[];
+  readonly eventPattern?: events.EventPattern;
+  readonly stateMachine: sfn.StateMachine;
+
+  /**
+   * Constructs EMRServerlessJobStage.
+   * @param scope Scope within which this construct is defined.
+   * @param id Identifier of the stage.
+   * @param props Properties for the stage.
+   */
+  constructor(scope: Construct, id: string, props: EMRServerlessJobStageProps) {
+    super(scope, id, props);
+
+    const stack = cdk.Stack.of(this);
+    const emrApplicationArn = `arn:${stack.partition}:emr-serverless:${stack.region}:${stack.account}:/applications/${props.applicationId}`;
+    const jobExecutionStatusWait = new sfn.Wait(this, "Wait Before Checking Job Status", {
+      time: sfn.WaitTime.duration(props.jobExecutionStatusCheckPeriod ?? Duration.seconds(30)),
+    });
+
+    const runJobTask = new tasks.CallAwsService(this, "Start Job Run", {
+      service: "emrserverless",
+      action: "startJobRun",
+      iamResources: [emrApplicationArn],
+      parameters: {
+        ApplicationId: props.applicationId,
+        ExecutionRoleArn: props.executionRoleArn,
+        JobDriver: props.jobDriver,
+        ClientToken: Math.random().toString(36),
+        ...props.startJobRunProps,
+      },
+    });
+
+    const getJobTask = new tasks.CallAwsService(scope, "Get Job Status", {
+      service: "emrserverless",
+      action: "getJobRun",
+      resultPath: "$.JobStatus",
+      iamResources: [emrApplicationArn],
+      parameters: {
+        "ApplicationId.$": "$.ApplicationId",
+        "JobRunId.$": "$.JobRunId",
+      },
+    });
+
+    const successState = new sfn.Succeed(this, "Success State");
+    const failState = new sfn.Fail(this, "Fail State");
+    const retryChain = jobExecutionStatusWait.next(getJobTask);
+    const jobStatusChoice = new sfn.Choice(scope, "Job Status Choice")
+      .when(sfn.Condition.stringEquals("$.JobStatus.JobRun.State", "SUCCESS"), successState)
+      .when(
+        sfn.Condition.or(
+          sfn.Condition.stringEquals("$.JobStatus.JobRun.State", "FAILED"),
+          sfn.Condition.stringEquals("$.JobStatus.JobRun.State", "CANCELLED"),
+        ),
+        failState,
+      )
+      .otherwise(retryChain);
+
+    const definition = runJobTask.next(getJobTask).next(jobStatusChoice);
+
+    ({
+      eventPattern: this.eventPattern,
+      targets: this.targets,
+      stateMachine: this.stateMachine,
+    } = this.createStateMachine({ definition: definition, ...props }));
+    this.stateMachine.addToRolePolicy(
+      new iam.PolicyStatement({
+        effect: iam.Effect.ALLOW,
+        actions: ["emr-serverless:StartJobRun", "emr-serverless:GetJobRun"],
+        resources: [emrApplicationArn, `${emrApplicationArn}/jobruns/*`],
+      }),
+    );
+    this.stateMachine.addToRolePolicy(
+      new iam.PolicyStatement({
+        effect: iam.Effect.ALLOW,
+        actions: ["iam:PassRole"],
+        resources: [props.executionRoleArn],
+      }),
+    );
+  }
+}
diff --git a/src/stages/index.ts b/src/stages/index.ts
index 3ed13f9d..3161ad40 100644
--- a/src/stages/index.ts
+++ b/src/stages/index.ts
@@ -1,6 +1,7 @@
 export * from "./appflow-ingestion";
 export * from "./athena-sql";
 export * from "./databrew-transform";
+export * from "./emr-serverless-job";
 export * from "./glue-transform";
 export * from "./kinesis-s3";
 export * from "./mwaa-trigger-dags";


Test patch:
diff --git a/test/emr-serverless-stage.test.ts b/test/emr-serverless-stage.test.ts
new file mode 100644
index 00000000..8e4d6df7
--- /dev/null
+++ b/test/emr-serverless-stage.test.ts
@@ -0,0 +1,54 @@
+import * as cdk from "aws-cdk-lib";
+import { Match, Template } from "aws-cdk-lib/assertions";
+import * as s3 from "aws-cdk-lib/aws-s3";
+
+import { EMRServerlessJobStage } from "../src";
+
+test("Basic Emr Stage", () => {
+  const stack = new cdk.Stack();
+
+  const bucket = new s3.Bucket(stack, "My Bucket");
+  new EMRServerlessJobStage(stack, "EMR Serverless Job", {
+    applicationId: "foobar",
+    executionRoleArn: "arn:aws:iam::1111111111111:role/EmrServerlessExecutionRole",
+    jobDriver: {
+      SparkSubmit: {
+        EntryPoint: `s3://${bucket.bucketName}/scripts/wordcount.py`,
+        EntryPointArguments: [`s3://${bucket.bucketName}/emr-serverless-spark/output`],
+        SparkSubmitParameters:
+          "--conf spark.executor.cores=1 --conf spark.executor.memory=4g --conf spark.driver.cores=1 --conf spark.driver.memory=4g --conf spark.executor.instances=1",
+      },
+    },
+  });
+  const template = Template.fromStack(stack);
+  template.hasResourceProperties("AWS::StepFunctions::StateMachine", {
+    DefinitionString: {
+      "Fn::Join": [
+        "",
+        Match.arrayWith([Match.stringLikeRegexp("Start Job Run"), Match.stringLikeRegexp("Get Job Status")]),
+      ],
+    },
+  });
+});
+
+test("Emr Stage Additional Parameters", () => {
+  const stack = new cdk.Stack();
+
+  const bucket = new s3.Bucket(stack, "My Bucket");
+  new EMRServerlessJobStage(stack, "EMR Serverless Job", {
+    applicationId: "foobar",
+    executionRoleArn: "arn:aws:iam::1111111111111:role/EmrServerlessExecutionRole",
+    jobExecutionStatusCheckPeriod: cdk.Duration.minutes(2),
+    jobDriver: {
+      SparkSubmit: {
+        EntryPoint: `s3://${bucket.bucketName}/scripts/wordcount.py`,
+        EntryPointArguments: [`s3://${bucket.bucketName}/emr-serverless-spark/output`],
+        SparkSubmitParameters:
+          "--conf spark.executor.cores=1 --conf spark.executor.memory=4g --conf spark.driver.cores=1 --conf spark.driver.memory=4g --conf spark.executor.instances=1",
+      },
+    },
+    startJobRunProps: {
+      Name: "foobar",
+    },
+  });
+});