AWS 步函数的任务

AWS CDK v1在2023-06-01上达到了支持结束. 此软件包不再更新,用户应该迁移到 AWS CDK v2.

关于如何移徙的更多信息,请参见: 移动到 AWS CDK v2 指南。 。 。 .

AWS 步函数 是一个网络服务,使您能够协调 使用视觉工作流程的分布式应用程序和微服务组件. 您从单个组件中构建应用程序,每个组件执行离散 函数,或任务,允许您快速缩放和更改应用程序.

A 类 任务 状态代表一个国家机器完成的单一工作单位. 你的国家机器里的所有工作都是通过任务完成的.

这个模块是 AWS 云开发套件 项目.

目录

• AWS 步函数的任务
  • 目录
  • 任务
  • 路径
    • 输入
    • 输出
    • 结果
  • 来自状态 JSON 的任务参数
  • 评估表达式
  • API 网关
    • 调用 REST API 端点
    • 调用 HTTP API 端点
  • 一个WS SDK
  • 雅典纳
    • 开始查询
    • 获取查询执行
    • 获取查询结果
    • 停止查询
  • 批次
    • 提交工作
  • 代码构建
    • 开始构建
  • 迪纳摩DB
    • 获取项目
    • 设置项目
    • 删除项目
    • 更新项目
  • 东加勒比国家
    • 运行任务
      • 欧共体2
      • 远端
  • EMR 电子数据交换系统
    • 创建集群
    • 终止保护
    • 终止集群
    • 添加步骤
    • 取消步骤
    • 修改实例组
    • 修改实例组
  • EKS上的EMR
    • 创建虚拟集群
    • 删除虚拟集群
    • 启动任务运行
  • 埃克萨斯
    • 打电话
  • 事件桥
    • 设置事件
  • 粘贴
  • 粘贴数据布鲁
  • 兰布达
  • 贤者
    • 创建培训任务
    • 创建变形任务
    • 创建终点
    • 创建端点配置
    • 创建模式
    • 更新终点
  • 简讯
  • 步骤函数
    • 开始执行
    • 启动活动
  • 质量

任务

任务状态(A Task state)代表一个国家机器完成的单个工作单位. 在那个 CDK,要完成的具体工作是由一个执行的类别决定的 IStepFunctionsTask。 。 。 .

AWS 步函数 整合 带有一些 AWS 服务以便您可以调用 API 并直接协调亚马逊州的处决 步函数. 您可以直接调用并传递参数到这些参数的 APIs 服务.

路径

亚马逊州语言,a 路径 是一个字符串开头 $ 那个你 可用于识别 JSON 文本中的组件.

更多了解 Step 函数中的输入和输出处理 这里

输入

两者 InputPath 财务报告和审定财务报表 Parameters 字段提供了操作 JSON 的方法 移动到工作流程中。 AWS 步函数应用 InputPath 字段,第一个, 然后是 Parameters 字段。 您可以先过滤您的原始输入到选择 ,然后应用参数来操纵输入 或添加新的值。 如果你没有指定一个 InputPath,默认值 联 合 国 $ 将被使用.

以下示例提供了命名的字段 input 作为输入 Task 用于运行 Lambda 函数的状态.

declare const fn: lambda.Function;
const submitJob = new tasks.LambdaInvoke(this, 'Invoke Handler', {
  lambdaFunction: fn,
  inputPath: '$.input',
});

输出

任务还允许您选择状态输出的一部分来传递到下一个 状态。 这使得您能够过滤不想要的信息, 并且只通过 你关心的JSON的一部分。 如果你没有指定一个 OutputPath, (中文(简体) ), 默认值为 $ 将被使用。 整个JSON节点都传到下一个 状态.

这个 回应 由 Lambda 函数生成的响应 以及其他元数据.

以下示例将任务输出指定为命名的字段 result

declare const fn: lambda.Function;
const submitJob = new tasks.LambdaInvoke(this, 'Invoke Handler', {
  lambdaFunction: fn,
  outputPath: '$.Payload.result',
});

结果选择器

你用得着 ResultSelector 操作任务、地图或并行状态的原始结果 转至 ResultPath。对于服务整合,原始 结果除了反应有效载荷外,还包含元数据。 你用得着 制造JSON有效载荷并成为有效结果的选择者 使用静态值或对原始结果或上下文对象的引用.

以下示例提取出 Lambda 函数的输出有效载荷 它带有一些静态值和来自上下文对象的状态名称.

declare const fn: lambda.Function;
new tasks.LambdaInvoke(this, 'Invoke Handler', {
  lambdaFunction: fn,
  resultSelector: {
    lambdaOutput: sfn.JsonPath.stringAt('$.Payload'),
    invokeRequestId: sfn.JsonPath.stringAt('$.SdkResponseMetadata.RequestId'),
    staticValue: {
      foo: 'bar',
    },
    stateName: sfn.JsonPath.stringAt('$$.State.Name'),
  },
});

结果

状态的输出可以是其输入的复制件,它所产生的结果(用于 例如,来自任务状态的 Lambda 函数的输出,或者其组合 投入和结果。 使用 ResultPath 控制其中哪一种组合 转到状态输出。 如果你没有指定一个 ResultPath,默认值 价值 $ 将被使用.

以下例子补充了拨打DynamotB's getItem API 到该州 输入并传递到下一个州.

declare const myTable: dynamodb.Table;
new tasks.DynamoPutItem(this, 'PutItem', {
  item: {
    MessageId: tasks.DynamoAttributeValue.fromString('message-id'),
  },
  table: myTable,
  resultPath: `$.Item`,
});

{\fn黑体\fs22\bord1\shad0\3aHBE\4aH00\fscx67\fscy66\2cHFFFFFF\3cH808080}你觉得呢 OutputPath 应用后计算 ResultPath. 所有服务整合 返回元数据作为其响应的一部分。 使用时 ResultPath这不可能 将任务输出的子集合并到输入中.

来自状态 JSON 的任务参数

大多数任务需要参数. 参数值可以是静态的,直接提供 在工作流程定义中(通过指定其值),或运行时可用的值 在状态机器的执行中(或者作为输入,或者作为先前状态的输出). 运行时可用的参数值可通过 JsonPath 班级, 使用方法,例如 JsonPath.stringAt()。 。 。 .

以下示例提供了命名的字段 input 作为 Lambda 函数的输入 并同时引用它.

declare const fn: lambda.Function;

const submitJob = new tasks.LambdaInvoke(this, 'Invoke Handler', {
  lambdaFunction: fn,
  payload: sfn.TaskInput.fromJsonPathAt('$.input'),
  invocationType: tasks.LambdaInvocationType.EVENT,
});

您还可以使用 内在函数 备查 JsonPath,例如 JsonPath.format()。 。 。 . 以下是启动雅典娜查询的例子, 使用任务输入来动态创建 :

const startQueryExecutionJob = new tasks.AthenaStartQueryExecution(this, 'Athena Start Query', {
  queryString: sfn.JsonPath.format('select contacts where year={};', sfn.JsonPath.stringAt('$.year')),
  queryExecutionContext: {
    databaseName: 'interactions',
  },
  resultConfiguration: {
    encryptionConfiguration: {
      encryptionOption: tasks.EncryptionOption.S3_MANAGED,
    },
    outputLocation: {
      bucketName: 'mybucket',
      objectKey: 'myprefix',
    },
  },
  integrationPattern: sfn.IntegrationPattern.RUN_JOB,
});

每个服务集成都有自己的一套可以提供的参数.

评估表达式

使用 EvaluateExpression 用于执行简单的操作,引用状态路径。 这个 expression 任务中的引用将在 Lambda 函数中进行评估 (单位:千美元)eval()) (中文(简体) ). 这使得您无需为简单的操作写出Lambda代码.

示例: 将等待时间由毫秒转换为秒, 将它汇入信件并等待 :

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_14_X,
  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),
});

这个 EvaluateExpression 支持 a runtime 指定 Lambda 的道具 用于评价表达式的运行时间。 目前,只有运行时间 支持Node.js家族.

API 网关

步骤函数支持 API 网关 通过服务整合模式.

HTTP API是为低纬度,成本效益高的与AWS服务的集成而设计的,包括AWS Lambda,和HTTP端点. HTTP API支持OIDC和OAuth 2.0授权,并随同内置支持CORS和自动部署. 上一代REST API目前提供更多的功能. 更多细节可以找到 这里。 。 。 .

调用 REST API 端点

这个 CallApiGatewayRestApiEndpoint 调用 REST API 端点.

import * as apigateway from '@aws-cdk/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,
});

注意信头值必须是数组. 传递任务时 在页眉字段中 WAIT_FOR_TASK_TOKEN 整合、使用 JsonPath.array() 以数组方式将符号包裹 :

import * as apigateway from '@aws-cdk/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),
  }),
});

调用 HTTP API 端点

这个 CallApiGatewayHttpApiEndpoint 调用 HTTP API 端点.

import * as apigatewayv2 from '@aws-cdk/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,
});

一个WS SDK

步骤函数支持调用 AWS 服务的 API 动作 通过服务整合模式.

您可以使用 Step Functions 的 AWS SDK 集成器调用任何超过 200 AWS 的服务 直接从你的国家机器, 让你进入九千多个API动作.

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('*')],
});

动作使用 CamamCase ,参数名称使用 PascalCase .

任务会自动添加一个IAM语句到国家机器角色的政策中 服务与行动。 此语句的资源必须在 iamResources。 。 。 .

使用 iamAction 在 IAM 情况下手动指定 IAM 动作名的道具 动作名不匹配 API 服务/ 动作名 :

const listBuckets = new tasks.CallAwsService(this, 'ListBuckets', {
  service: 's3',
  action: 'listBuckets',
  iamResources: ['*'],
  iamAction: 's3:ListAllMyBuckets',
});

雅典纳

步骤函数支持 雅典纳 通过服务整合模式.

开始查询

这个 开始查询 API运行SQL查询语句.

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',
    },
  },
});

获取查询执行

这个 获取查询执行 API获得一个查询的单个执行信息.

const getQueryExecutionJob = new tasks.AthenaGetQueryExecution(this, 'Get Query Execution', {
  queryExecutionId: sfn.JsonPath.stringAt('$.QueryExecutionId'),
});

获取查询结果

这个 获取查询结果 API 从 S3 流出由 QueryExecutionId 指定的单一查询执行的结果 .

const getQueryResultsJob = new tasks.AthenaGetQueryResults(this, 'Get Query Results', {
  queryExecutionId: sfn.JsonPath.stringAt('$.QueryExecutionId'),
});

停止查询

这个 停止查询 API 停止查询执行 .

const stopQueryExecutionJob = new tasks.AthenaStopQueryExecution(this, 'Stop Query Execution', {
  queryExecutionId: sfn.JsonPath.stringAt('$.QueryExecutionId'),
});

批次

步骤函数支持 批次 通过服务整合模式.

提交工作

这个 提交工作 API 从任务定义中提交 AWS 批次任务 .

import * as batch from '@aws-cdk/aws-batch';
declare const batchJobDefinition: batch.JobDefinition;
declare const batchQueue: batch.JobQueue;

const task = new tasks.BatchSubmitJob(this, 'Submit Job', {
  jobDefinitionArn: batchJobDefinition.jobDefinitionArn,
  jobName: 'MyJob',
  jobQueueArn: batchQueue.jobQueueArn,
});

代码构建

步骤函数支持 代码构建 通过服务整合模式.

开始构建

开始构建 按工程名称启动一个代码构建工程.

import * as codebuild from '@aws-cdk/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'),
    },
  },
});

迪纳摩DB

您可以从一个 Task 状态. 更多信息,请访问DynamoDB APIs 这里

获取项目

这个 获取项目 操作返回带有给定的主键的项目的一组属性.

declare const myTable: dynamodb.Table;
new tasks.DynamoGetItem(this, 'Get Item', {
  key: { messageId: tasks.DynamoAttributeValue.fromString('message-007') },
  table: myTable,
});

设置项目

这个 设置项目 操作创建新项目,或用新项目取代旧项目.

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,
});

删除项目

这个 删除项目 操作按主键删除表格中的单个项目.

declare const myTable: dynamodb.Table;
new tasks.DynamoDeleteItem(this, 'DeleteItem', {
  key: { MessageId: tasks.DynamoAttributeValue.fromString('message-007') },
  table: myTable,
  resultPath: sfn.JsonPath.DISCARD,
});

更新项目

这个 更新项目 操作编辑已存在的项目属性,或添加新项目 如果它不存在,则将其移到表格中.

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',
});

东加勒比国家

步骤函数支持 ECS/远门 通过服务整合模式.

运行任务

运行任务 使用指定的任务定义启动新任务.

欧共体2

EC2 发射类型允许您在一个集群上运行您的容器化应用程序 您管理 Amazon EC2 实例.

当使用EC2发射类型的任务被启动时,亚马逊ECS必须确定在哪里 以便根据任务定义中规定的要求安排任务,例如: CPU和内存. 同样,当你缩小任务计数时,亚马逊ECS必须确定 任务终止。 您可运用任务安排策略和限制 自定义 Amazon ECS 如何设置和终止任务 。 学习更多关于 任务定位

最新的ACTIVE修订已通过的任务定义用于任务运行.

以下实例从EC2的任务定义中运行一个任务

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'),
    ],
  }),
});

远端

AWS Fargate 是用于与亚马逊公司合作的容器的无服务器计算引擎 弹性集装箱服务。 远门方便你专心建设 您的应用程序。 Fargate 取消了提供和管理服务器的需要, 允许您使用 每个应用程序指定和支付资源,并通过应用程序加强安全 设计上的隔离。 学习更多关于 远端

Fargate 发射类型允许您无需操作的容器化应用程序 提供和管理后端基础设施。 只要登记你的任务定义和 远门为你发射容器。 最新修订 任务定义用于任务运行。 学习更多关于 远端版本

以下示例从 Fargate 的任务定义中运行一个任务

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(),
});

EMR 电子数据交换系统

Step函数通过服务整合模式支持亚马逊EMR. 服务集成 API 对应 Amazon EMR API ,但不同于 参数.

更多信息 关于使用这些服务集成时的差异.

创建集群

创建并开始运行集群(工作流程). 对应 runJobFlow API in EMR. 互联网档案馆的存檔,存档日期2013-12-02.

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,
});

如果您想要运行多个步骤 平行, (中文(简体) ), 您可以指定 stepConcurrencyLevel 属性。 货币范围在 1 之间 和256 兼容并包,其中1的默认货币表示不允许步元. stepConcurrencyLevel 要求EMR释放标签为5.2.8.0或以上.

new tasks.EmrCreateCluster(this, 'Create Cluster', {
  instances: {},
  name: sfn.TaskInput.fromJsonPathAt('$.ClusterName').value,
  stepConcurrencyLevel: 10,
});

终止保护

锁定集群( 工作流程) , 因此集群中的 EC2 实例不能是 通过用户干预、API呼叫或工作流程出错而终止.

对应 setTerminationProtection API in EMR. 互联网档案馆的存檔,存档日期2013-12-02.

new tasks.EmrSetClusterTerminationProtection(this, 'Task', {
  clusterId: 'ClusterId',
  terminationProtected: false,
});

终止集群

关闭集群( 工作流量) . 对应 terminateJobFlows API in EMR. 互联网档案馆的存檔,存档日期2013-12-02.

new tasks.EmrTerminateCluster(this, 'Task', {
  clusterId: 'ClusterId',
});

添加步骤

在运行的集群中添加新步骤. 对应 addJobFlowSteps API in EMR. 互联网档案馆的存檔,存档日期2013-12-02.

new tasks.EmrAddStep(this, 'Task', {
  clusterId: 'ClusterId',
  name: 'StepName',
  jar: 'Jar',
  actionOnFailure: tasks.ActionOnFailure.CONTINUE,
});

取消步骤

取消运行中的集群中的待决步骤. 对应 cancelSteps API in EMR. 互联网档案馆的存檔,存档日期2013-12-02.

new tasks.EmrCancelStep(this, 'Task', {
  clusterId: 'ClusterId',
  stepId: 'StepId',
});

修改实例组

修改目标 " 待命 " 和 " 点 " 能力 与指定的 EntextFleet 的舰队Name.

对应 modifyInstanceFleet API in EMR. 互联网档案馆的存檔,存档日期2013-12-02.

new tasks.EmrModifyInstanceFleetByName(this, 'Task', {
  clusterId: 'ClusterId',
  instanceFleetName: 'InstanceFleetName',
  targetOnDemandCapacity: 2,
  targetSpotCapacity: 0,
});

修改实例组

修改实例组的节点数和配置设置.

对应 modifyInstanceGroups API in EMR. 互联网档案馆的存檔,存档日期2013-12-02.

new tasks.EmrModifyInstanceGroupByName(this, 'Task', {
  clusterId: 'ClusterId',
  instanceGroupName: sfn.JsonPath.stringAt('$.InstanceGroupName'),
  instanceGroup: {
    instanceCount: 1,
  },
});

EKS上的EMR

Step Functions通过服务集成模式支持在EKS上的AmazonEMR. 服务集成API在EKS API上对应了亚马逊EMR,但使用的参数不同.

更多信息 关于使用这些服务集成时的差异.

设置 需要EKS集群.

创建虚拟集群

这个 创建虚拟组合 API创建一个被映射到单一Kubernetes命名空间的单一虚拟集群.

包含 Kubernetes 命名空间的 EKS 集群, 虚拟集群将被映射到其中, 可以从任务输入中传递 .

new tasks.EmrContainersCreateVirtualCluster(this, 'Create a Virtual Cluster', {
  eksCluster: tasks.EksClusterInput.fromTaskInput(sfn.TaskInput.fromText('clusterId')),
});

EKS集群也可以直接通过.

import * as eks from '@aws-cdk/aws-eks';

declare const eksCluster: eks.Cluster;

new tasks.EmrContainersCreateVirtualCluster(this, 'Create a Virtual Cluster', {
  eksCluster: tasks.EksClusterInput.fromCluster(eksCluster),
});

默认情况下,Kubernetes命名空间,一个虚拟集群地图是"默认",但可以选择一个EKS集群内的特定命名空间.

new tasks.EmrContainersCreateVirtualCluster(this, 'Create a Virtual Cluster', {
  eksCluster: tasks.EksClusterInput.fromTaskInput(sfn.TaskInput.fromText('clusterId')),
  eksNamespace: 'specified-namespace',
});

删除虚拟集群

这个 删除虚拟组合 API删除虚拟集群.

new tasks.EmrContainersDeleteVirtualCluster(this, 'Delete a Virtual Cluster', {
  virtualClusterId: sfn.TaskInput.fromJsonPathAt('$.virtualCluster'),
});

启动任务运行

这个 开始JobRun API开始工作运行. 一个工作是一个工作单位,你提交亚马逊EMR在EKS上执行. 任务所完成的工作可以通过 Spark 罐, PySpark 脚本,或 SparkSQL 查询来定义. 任务运行是在虚拟集群上执行任务.

需要设置 :

• 如果还没有完成,请跟随 步骤 设置 EKS 上的 EMR 和 创建 EKS 集群。 。 。 .
• 启用 集群接入
• 启用 IAM 角色访问

如果从任务输入中提供虚拟集群ID,则必须执行以下动作. 否则,如果在状态机器定义中静态地提供,这些行动会自动完成.

• 创建一个 IAM 角色
• 更新 角色信托政策 执行职务.

任务可以用火花提交参数配置 :

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',
    },
  },
});

配置任务也可以通过应用程序配置完成:

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',
    },
  }],
});

如果 monitoring.logging 被设定为真实。 这自动生成 S3 桶和 Cloud Watch 日志 .

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,
  },
});

否则,还可用现有的日志组和日志桶对工作进行监测.

import * as logs from '@aws-cdk/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,
  },
});

用户可以提供自己现有的工作执行角色.

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',
    },
  },
});

埃克萨斯

Step函数通过服务整合模式支持亚马逊EKS. 服务集成API对应于亚马逊 EKS API.

更多信息 关于使用这些服务集成时的差异.

打电话

通过 Kubernetes API 端点读取并写入 Kubernetes 资源对象 . 对应 call Step 函数连接器中的 API .

以下代码片段包括一个使用 eks: call 的 Task 状态来列出 pock.

import * as eks from '@aws-cdk/aws-eks';

const myEksCluster = new eks.Cluster(this, 'my sample cluster', {
  version: eks.KubernetesVersion.V1_18,
  clusterName: 'myEksCluster',
});

new tasks.EksCall(this, 'Call a EKS Endpoint', {
  cluster: myEksCluster,
  httpMethod: tasks.HttpMethods.GET,
  httpPath: '/api/v1/namespaces/default/pods',
});

事件桥

Step函数通过服务整合模式支持亚马逊事件Bridge. 服务集成API对应亚马逊事件Bridge API.

更多信息 关于使用这些服务集成时的差异.

设置事件

将事件发送到 EventBridge 公共汽车. 对应 put-events Step 函数连接器中的 API .

以下代码片段包括使用事件的任务状态: putevents将事件发送到默认总线.

import * as events from '@aws-cdk/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',
  }],
});

粘贴

步骤函数支持 AWS 粘贴 通过服务整合模式.

你可以打电话给 StartJobRun API 来自 a Task 状态.

new tasks.GlueStartJobRun(this, 'Task', {
  glueJobName: 'my-glue-job',
  arguments: sfn.TaskInput.fromObject({
    key: 'value',
  }),
  timeout: Duration.minutes(30),
  notifyDelayAfter: Duration.minutes(5),
});

粘贴数据布鲁

步骤函数支持 AWS 粘贴数据组 通过服务整合模式.

你可以打电话给 StartJobRun API 来自 a Task 状态.

new tasks.GlueDataBrewStartJobRun(this, 'Task', {
  name: 'databrew-job',
});

兰布达

调用 a Lambda函数.

您可以通过 payload 属性. 默认情况下, Step Functions 引用 Lambda 函数并附带状态输入( JSON path '$') 作为输入.

以下片段引用带有状态输入的 Lambda 函数作为有效载荷 通过引用 $ 路径.

declare const fn: lambda.Function;
new tasks.LambdaInvoke(this, 'Invoke with state input', {
  lambdaFunction: fn,
});

当一个函数被引用时, Lambda 服务发送 这些反应 要素 回来.

Lambda 函数的反应在一个属性中称为 Payload

以下片段引用了 Lambda 函数,引用了 $.Payload 路径 引用在其前执行的 Lambda 的输出.

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'),
});

以下片段引用了 Lambda 并设定任务输出只包含 lambda 函数响应.

declare const fn: lambda.Function;
new tasks.LambdaInvoke(this, 'Invoke and set function response as task output', {
  lambdaFunction: fn,
  outputPath: '$.Payload',
});

如果您想要合并输入和 Lambda 函数响应, 您可以使用 页:1 payloadResponseOnly 属性并指定 resultPath。这样可以让 Lambda在"资源"字符串中直接发挥ARN的功能,但它与"资源"的RN冲突 融合 图案,引用Type,客户端Context,以及限定属性.

declare const fn: lambda.Function;
new tasks.LambdaInvoke(this, 'Invoke and combine function response with task input', {
  lambdaFunction: fn,
  payloadResponseOnly: true,
  resultPath: '$.fn',
});

您可以让 Step 函数暂停一个任务, 等待外部进程到 返回任务符号。 更多关于 回调模式

要使用回调模式,请设置 token 属性。 调用步骤 职能 SendTaskSuccess 或者说 SendTaskFailure 带有符号的 API 说明任务已经完成,国家机器应恢复执行.

以下片段引用带有任务符号的 Lambda 作为输入的一部分 去兰巴达.

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'),
  }),
});

任务将暂停,直到它收到以一个 SendTaskSuccess 或者说 SendTaskFailure 电话 学习更多关于 随任务召回 脚本。 。 。 .

AWS Lambda偶尔会遇到短暂的服务错误. 在此案中,援引Lambda 导致500个错误,例如 ServiceException, (中文(简体) ), AWSLambdaException,或 SdkClientException。 。 。 . 作为最佳做法 LambdaInvoke 任务将重试这些错误,间隔为2秒, 退后率2和6 设定 retryOnServiceExceptions 向 false 改为 禁用此行为 .

贤者

步骤函数支持 AWS 赛格麦克 通过服务整合模式.

如果你的训练工作或模型使用来自AWS 市场的资源, 需要网络隔离。 。 。 . 为此,设置 enableNetworkIsolation 属性改为 true 页:1 SageMakerCreateModel 或者说 SageMakerCreateTrainingJob。 。 。 .

要设置 Docker 容器的环境变量,请使用 environment 属性.

创建培训任务

你可以打电话给 CreateTrainingJob API 来自 a Task 状态.

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', 'mybucket'), '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
});

创建变形任务

你可以打电话给 CreateTransformJob API 来自 a Task 状态.

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),
  }
});

创建终点

你可以打电话给 CreateEndpoint API 来自 a Task 状态.

new tasks.SageMakerCreateEndpoint(this, 'SagemakerEndpoint', {
  endpointName: sfn.JsonPath.stringAt('$.EndpointName'),
  endpointConfigName: sfn.JsonPath.stringAt('$.EndpointConfigName'),
});

创建端点配置

你可以打电话给 CreateEndpointConfig API 来自 a Task 状态.

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',
  }],
});

创建模式

你可以打电话给 CreateModel API 来自 a Task 状态.

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'),
  }),
});

更新终点

你可以打电话给 UpdateEndpoint API 来自 a Task 状态.

new tasks.SageMakerUpdateEndpoint(this, 'SagemakerEndpoint', {
  endpointName: sfn.JsonPath.stringAt('$.Endpoint.Name'),
  endpointConfigName: sfn.JsonPath.stringAt('$.Endpoint.EndpointConfig'),
});

简讯

步骤函数支持 亚马逊 SNS 通过服务整合模式.

你可以打电话给 Publish API 来自 a Task 状态将发布到 SNS 主题.

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'),
  }),
});

步骤函数

开始执行

你可以处理 AWS 步函数 处决.

AWS 步函数支持它自己的 StartExecution API作为服务集成.

// 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,
});

你可以利用 相关工作流程执行 通过 associateWithParent 属性。 这允许 Step 函数 UI 连接子 通过父母处决处决,更容易追踪国家机器之间的处决流.

declare const child: sfn.StateMachine;
const task = new tasks.StepFunctionsStartExecution(this, 'ChildTask', {
  stateMachine: child,
  associateWithParent: true,
});

这将增加有效载荷 AWS_STEP_FUNCTIONS_STARTED_BY_EXECUTION_ID.$: $$.Execution.Id 页:1 input属性,它将将执行 ID 从上下文对象传递到 执行输入。 这需要 input 要成为对象或根本不被设定.

启动活动

你可以援引 步骤函数活动 让你拥有 工作由下列人员完成的状态机器中的任务 劳动者 这可以 在亚马逊EC2,亚马逊ECS,AWS Lambda上主机,基本上在任何地方. 活 动 是一种将代码运行于某地(被称为活动工作者)与 国家机器中的一项具体任务.

当 Step 函数达到活动任务状态时,工作流程会等待一个 活动工作者为某项任务进行投票。 工作统计 使用 Get 活动 任务,并发送 ARN 用于相关活动 .

劳动劳动者完成劳动后,可以提供劳动状况报告 使用成功或失败 SendTaskSuccess 或者说 SendTaskFailure这两位 调用 GetActivity 任务提供的任务托肯来关联结果 与这个任务.

以下例子创造了一项活动,并创造了一项引用该活动的任务.

const submitJobActivity = new sfn.Activity(this, 'SubmitJob');

new tasks.StepFunctionsInvokeActivity(this, 'Submit Job', {
  activity: submitJobActivity,
});

质量

步骤函数支持 亚马逊 SQS

你可以打电话给 SendMessage API 来自 a Task 状态 向 SQS 队列发送消息.

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'),
  }),
});