云工作流：分布式模式

基本概念

分布式模式：迭代（Map）状态的一种处理模式，对OSS数据进行迭代，允许高并发处理。该模式支持从对象存储Bucket中的对象列表或文件作为数据来源。

子工作流执行：当迭代（Map）状态被设置为分布式模式时，Map迭代的执行会以子工作流执行的形式进行，此时子工作流的定义为Map Processor的状态转换定义，接收的输入为Map迭代元素。MapRun子流程执行模式支持快速模式和标准模式，不同模式的区别，请参见标准工作流和快速工作流。

故障容忍策略：对于Map执行，当子工作流执行失败时，Map状态会以失败状态结束。对于迭代数量较多的Map执行，可以通过配置错误容忍使Map执行在子工作流执行失败时继续执行。

分布式模式包含以下字段：

使用限制

配额限制

对于分布式模式执行存在以下配额限制，如果默认配额不能满足您的需求，您可以通过工单提交配额提升申请。

角色限制

对于分布式模式的执行，您需要具备以下角色权限。有关创建角色权限的详细信息，请参见创建权限策略。

{
    "Version": "1",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "oss:HeadObject",
                "oss:GetObjectMeta",
                "oss:GetObject",
                "oss:PutObject",
                "oss:ListObjectsV2",
                "oss:ListObjects",
                "oss:InitiateMultipartUpload",
                "oss:UploadPart",
                "oss:CompleteMultipartUpload",
                "oss:AbortMultipartUpload",
                "oss:ListMultipartUploads",
                "oss:ListParts"
            ],
            "Resource": "*"
        }
    ]
}

分布式模式的关键字段

ItemsPath

用于提取输入数组的表达式。该表达式执行后返回JSON Array，则可以进行迭代，将其中每个元素传入ItemProcessor进行处理；可使用表达式变量$Context和$Input，示例如下：

$Input.FieldA

Processor

迭代处理器。Processor包含的字段如下表所示。

Processor:
   StartAt: Pass1
   States:
     - Type: Pass
       Name: Pass1
       End: true

ProcessorConfig

处理器配置。ProcessorConfig包含的字段如下表所示。

ItemReader

Map分布式模式通过支持从OSS读取数据输入的方式，支持更大的数据输入。ItemReader包含的字段如下表所示。

SourceParameters

来源参数。SourceParameters包含的字段如下表所示。

ReaderConfig

阅读器配置。ReaderConfig包含的字段如下表所示。

从OSS读取数据的方式包括以下几个方法：

• 存储于OSS中的CSV文件

  使用Map状态从一个CSV文件中读取数据。假设您有一个example-object.csv文件存储在OSS的存储桶example-bucket中，可参考以下状态机定义示例。

  Type: StateMachine
  Name: MyWorkflow
  SpecVersion: v1
  StartAt: Map
  States:
    - Type: Map
      Name: Map
      ProcessorConfig:
        ExecutionMode: Express
      Processor:
        StartAt: Pass
        States:
          - Type: Pass
            Name: Pass
            End: true
      ItemReader:
        SourceType: OSS_CSV
        SourceParameters:
          Bucket: example-bucket
          ObjectName: example-object.csv
      End: true

  以下示例显示了example-object.csv文件的内容。

  ColA,ColB,ColC
  col_a_1,col_b_1,col_c_1

  以下示例显示了子工作流执行接收的输入。

  {
    "ColA": "col_a_1",
    "ColB": "col_b_1",
    "ColC": "col_c_1",
  }

• 存储于OSS中的JSON文件

  重要

  存储于OSS中的JSON文件所包含的JSON内容必须是一个JSON Array。

  用于从指定的OSS中读取JSON列表。假设您有一个example-object.json文件存储在OSS存储桶example-bucket中，可参考以下状态机定义示例。

  Type: StateMachine
  Name: MyWorkflow
  SpecVersion: v1
  StartAt: Map
  States:
    - Type: Map
      Name: Map
      ProcessorConfig:
        ExecutionMode: Express
      Processor:
        StartAt: Pass
        States:
          - Type: Pass
            Name: Pass
            End: true
      ItemReader:
        SourceType: OSS_JSON_LIST
        SourceParameters:
          Bucket: example-bucket
          ObjectName: example-object.json
      End: true

  以下示例显示了example-object.json文件的内容。

  [
    {
      "key_1": "value_1"
    }
  ]

  以下示例显示了子工作流执行接收的输入。

  {
    "key_1": "value_1",
  }

• 列举OSS中的Objects

  用于从指定的OSS中读取对象。假设您已将Objects存储在OSS存储桶example-bucket中，命名的前缀为example-prefix，可参考以下状态机定义示例。

  Type: StateMachine
  Name: MyWorkflow
  SpecVersion: v1
  StartAt: Map
  States:
    - Type: Map
      Name: Map
      ProcessorConfig:
        ExecutionMode: Express
      Processor:
        StartAt: Pass
        States:
          - Type: Pass
            Name: Pass
            End: true
      ItemReader:
        SourceType: OSS_OBJECTS
        SourceParameters:
          Bucket: example-bucket
          Prefix: example-prefix
      End: true

  以下目录结构示意图表明example-prefix/object_1文件存储在OSS存储桶example-bucket中。

  example-bucket
     ├── example-prefix/object_1

  以下示例显示了子工作流执行接收的输入。

  {
    "XMLName": {
      "Space": "",
      "Local": "Contents"
    },
    "Key": "example-prefix/object_1",
    "Type": "Normal",
    "Size": 268435,
    "ETag": "\"50B06D6680D86F04138HSN612EF5DEC6\"",
    "Owner": {
      "XMLName": {
        "Space": "",
        "Local": ""
      },
      "ID": "",
      "DisplayName": ""
    },
    "LastModified": "2024-01-01T01:01:01Z",
    "StorageClass": "Standard",
    "RestoreInfo": ""
  }

• 存储于OSS中的Inventory

  使用Map状态从指定的OSS存储桶和文件中读取数据。假设您已将清单文件manifest.json存储在example-bucket的OSS存储桶中，路径是inventory/2024-01-01T01-01Z/manifest.json，可参考以下状态机定义示例。

  Type: StateMachine
  Name: MyWorkflow
  SpecVersion: v1
  StartAt: Map
  States:
    - Type: Map
      Name: Map
      ProcessorConfig:
        ExecutionMode: Express
      Processor:
        StartAt: Pass
        States:
          - Type: Pass
            Name: Pass
            End: true
      ItemReader:
        SourceType: OSS_INVENTORY_FILES
        SourceParameters:
          Bucket: example-bucket
          ObjectName: inventory/2024-01-01T01-01Z/manifest.json
      ItemConstructor:
        Key.$: $Item.Key
      End: true

  以下示例显示了清单文件的内容。

  "example-bucket","object_name_1"  
  "example-bucket","object_name_2"

  以下示例显示第一个子工作流执行接收的输入。

  {
    "Bucket": "example-bucket",
    "Key": "object_name_1"
  }

ItemBatcher

开启ItemBatcher可以将多个Item聚合为Item Batch作为输入传递给子工作流执行。ItemBatcher包含的字段如下表所示。

MaxItemsPerBatch

MaxItemsPerBatch支持按照数量对Item进行聚合。以下是MaxItemsPerBatch的使用示例。

Type: StateMachine
Name: MyWorkflow
SpecVersion: v1
StartAt: Map
States:
  - Type: Map
    Name: Map
    ProcessorConfig:
      ExecutionMode: Express
    ItemsPath: $Input.Items
    Processor:
      StartAt: Pass
      States:
        - Type: Pass
          Name: Pass
          End: true
    ItemBatcher:
      MaxItemsPerBatch: 2
    End: true

以下示例是状态机执行的输入。

{
  "Items": [
    {"key_1":"value_1"},
    {"key_2":"value_2"},
    {"key_3":"value_3"},
    {"key_4":"value_4"},
    {"key_5":"value_5"}
  ]
}

以下示例显示了子工作流执行接收的输入。

# execution-1
# 第1个子工作流输入示例
{
  "Items": [
    {"key_1":"value_1"},
    {"key_2":"value_2"}
  ]
}

# execution-2
# 第2个子工作流输入示例
{
  "Items": [
    {"key_1":"value_3"},
    {"key_2":"value_4"}
  ]
}

# execution-3
# 第3个子工作流输入示例
{
  "Items": [
    {"key_1":"value_5"},
  ]
}

MaxInputBytesPerBatch

MaxInputBytesPerBatch支持按照聚合后输入的大小对Item进行聚合，MaxInputBytesPerBatch可以确保聚合后的输入不超过指定的大小。以下是MaxInputBytesPerBatch的使用示例。

Type: StateMachine
Name: MyWorkflow
SpecVersion: v1
StartAt: Map
States:
  - Type: Map
    Name: Map
    ProcessorConfig:
      ExecutionMode: Express
    ItemsPath: $Input.Items
    Processor:
      StartAt: Pass
      States:
        - Type: Pass
          Name: Pass
          End: true
    ItemBatcher:
      MaxInputBytesPerBatch: 50
    End: true

以下示例是状态机执行的输入。

{
  "Items":[
    {"Key":1},
    {"key":2},
    {"Key":3},
    {"Key":4},
    {"Key":5}
  ]
}

以下示例显示了子工作流执行接收的输入。

# execution-1
# 第1个子工作流输入示例
{
  "Items":[
    {"Key":1},
    {"key":2}
  ]
}

# execution-2
# 第2个子工作流输入示例
{
  "Items":[
    {"Key":3},
    {"key":4}
  ]
}

# execution-3
# 第3个子工作流输入示例
{
  "Items":[
    {"Key":5}
  ]
}

BatchInput

BatchInput支持在聚合Items时添加额外的输入。以下是BatchInput的使用示例。

Type: StateMachine
Name: MyWorkflow
SpecVersion: v1
StartAt: Map
States:
  - Type: Map
    Name: Map
    ProcessorConfig:
      ExecutionMode: Express
    ItemsPath: $Input.Items
    Processor:
      StartAt: Pass
      States:
        - Type: Pass
          Name: Pass
          End: true
    ItemBatcher:
      MaxInputBytesPerBatch: 70
      BatchInput:
        InputKey.$: $Input.Key
    End: true

以下示例是状态机执行的输入。

{
  "Key":"value",
  "Items":[
    {"Key":1},
    {"key":2},
    {"Key":3},
    {"Key":4},
    {"Key":5}
  ]
}

以下示例显示了子工作流执行接收的输入。

# execution-1
# 第1个子工作流输入示例
{
  "BatchInput":{
    "InputKey":"value"
  },
  "Items":[
    {"Key":1},
    {"key":2}
  ]
}

# execution-2
# 第2个子工作流输入示例
{
  "BatchInput":{
    "InputKey":"value"
  },
  "Items":[
    {"Key":3},
    {"key":4}
  ]
}

# execution-3
# 第3个子工作流输入示例
{
  "BatchInput":{
    "InputKey":"value"
  },
  "Items":[
    {"Key":5}
  ]
}

ItemConstructor

ItemConstructor支持对Item进行单独构造，ItemConstructor支持通过$Item引用要被构造的Item。以下是ItemConstructor的使用示例。

Type: StateMachine
Name: MyWorkflow
SpecVersion: v1
StartAt: Map
States:
  - Type: Map
    Name: Map
    ProcessorConfig:
      ExecutionMode: Express
    ItemsPath: $Input.Items
    Processor:
      StartAt: Pass
      States:
        - Type: Pass
          Name: Pass
          End: true
    ItemBatcher:
      MaxInputBytesPerBatch: 200
      BatchInput:
        InputKey.$: $Input.Key
    ItemConstructor:
      ConstructedKey.$: $Item.Key
      InputKey.$: $Input.Key
    End: true

以下示例是状态机执行的输入。

{
  "Key":"value",
  "Items":[
    {"Key":1},
    {"Key":2},
    {"Key":3},
    {"Key":4},
    {"Key":5}
  ]
}

以下示例显示了子工作流执行接收的输入。

# execution-1
# 第1个子工作流输入示例
{
  "BatchInput": {
    "InputKey": "value"
  },
  "Items": [
    {
      "InputKey": "value",
      "ConstructedKey": 1
    },
    {
      "InputKey": "value",
      "ConstructedKey": 2
    },
    {
      "InputKey": "value",
      "ConstructedKey": 3
    }
  ]
}

# execution-2
# 第2个子工作流输入示例
{
  "BatchInput": {
    "InputKey": "value"
  },
  "Items": [
    {
      "InputKey": "value",
      "ConstructedKey": 4
    },
    {
      "InputKey": "value",
      "ConstructedKey": 5
    }
  ]
}

ResultWriter

ResultWriter支持将子工作流执行的信息写入到指定的OSS目录。ResultWriter包含的字段如下表所示。

Parameters

请求参数。Parameters包含的字段如下表所示。

以下是ResultWriter的使用示例。

Type: StateMachine
Name: MyWorkflow
SpecVersion: v1
StartAt: Map
States:
  - Type: Map
    Name: Map
    ProcessorConfig:
      ExecutionMode: Express
    ItemsPath: $Input.Items
    ItemConstructor:
      Key.$: $Item.Key
      FailedValue.$: $Input.FailedValue
    ToleratedFailurePercentage: 30
    Processor:
      StartAt: Choice
      States:
        - Type: Choice
          Name: Choice
          Branches:
            - Condition: $Input.Key > $Input.FailedValue
              Next: Fail
          Default: Succeed
        - Type: Succeed
          Name: Succeed
          End: true
        - Type: Fail
          Name: Fail
          Code: MockError
          End: true
    ResultWriter:
      Parameters:
        Bucket: example-bucket
        Prefix: example-prefix/
    End: true

以下示例是状态机执行的输入。

{
  "FailedValue": 4,
  "Items": [
    {
      "Key": 1
    },
    {
      "Key": 2
    },
    {
      "Key": 3
    },
    {
      "Key": 4
    },
    {
      "Key": 5
    }
  ]
}

以下示例显示了3个JSON文件内容。

# 以下是manifest.json文件的内容，存储路径为example-prefix/map-run-name/manifest.json
{
    "DestinationBucket": "example-bucket",
    "MapRunName": "map-run-name",
    "ResultFiles": {
        "FAILED": [
            {
                "ObjectName": "example-prefix/map-run-name/FAILED_0.json",
                "Size": 262
            }
        ],
        "SUCCEED": [
            {
                "ObjectName": "example-prefix/map-run-name/SUCCEED_0.json",
                "Size": 1057
            }
        ]
    }
}

# 以下是FAILED_0.json文件的内容，存储路径为example-prefix/map-run-name/FAILED_0.json
[
    {
        "ExecutionName": "execution-name-5",
        "FlowName": "example",
        "Input": "{\"FailedValue\":4,\"Key\":5}",
        "Output": "{\"ErrorCode\":\"MockError\"}",
        "Status": "Failed",
        "StartedTime": "rfc3339-format-time-string",
        "StoppedTime": "rfc3339-format-time-string"
    }
]

# 以下是SUCCEED_0.json文件的内容，存储路径为example-prefix/map-run-name/SUCCEED_0.json
[
    {
        "ExecutionName": "execution-name-1",
        "FlowName": "example",
        "Input": "{\"FailedValue\":4,\"Key\":1}",
        "Output": "{\"FailedValue\":4,\"Key\":1}",
        "Status": "Succeeded",
        "StartedTime": "rfc3339-format-time-string",
        "StoppedTime": "rfc3339-format-time-string"
    },
    {
        "ExecutionName": "execution-name-2",
        "FlowName": "example",
        "Input": "{\"FailedValue\":4,\"Key\":2}",
        "Output": "{\"FailedValue\":4,\"Key\":2}",
        "Status": "Succeeded",
        "StartedTime": "rfc3339-format-time-string",
        "StoppedTime": "rfc3339-format-time-string"
    },
    {
        "ExecutionName": "execution-name-3",
        "FlowName": "example",
        "Input": "{\"FailedValue\":4,\"Key\":3}",
        "Output": "{\"FailedValue\":4,\"Key\":3}",
        "Status": "Succeeded",
        "StartedTime": "rfc3339-format-time-string",
        "StoppedTime": "rfc3339-format-time-string"
    },
    {
        "ExecutionName": "execution-name-4",
        "FlowName": "example",
        "Input": "{\"FailedValue\":4,\"Key\":4}",
        "Output": "{\"FailedValue\":4,\"Key\":4}",
        "Status": "Succeeded",
        "StartedTime": "rfc3339-format-time-string",
        "StoppedTime": "rfc3339-format-time-string"
    }
]

MaxItems

MaxItems支持配置Map最多处理的Item数量。例如对于包含了10000个Object的特定OSS Prefix，如果配置MaxItems为1000，则Map只会从OSS加载1000个Object。

MaxConcurrency

MaxConcurrency支持配置子工作流执行并发。例如对于包含了10000个Item的Map执行，当MaxConcurrency配置为100时，Map同时最多执行100个子工作流。

ToleratedFailurePercentage

ToleratedFailurePercentage支持按照百分比配置失败容忍策略，例如当Item总数为10000时，配置ToleratedFailurePercentage为10可以使 Map 容忍最多 1000 个 Item 处理失败。

ToleratedFailureCount

ToleratedFailureCount支持按照数量配置失败容忍策略，例如当Item总数为10000时，配置ToleratedFailureCount为10可以使Map容忍最多10个Item处理失败。

使用示例

如下工作流定义包含了一个分布式模式的Map状态，它从上游状态读取输入，并通过$Input.Items提取迭代元素。对于每一个迭代元素，Map会创建一个Express模式的子工作流执行。

Type: StateMachine
Name: MyWorkflow
SpecVersion: v1
StartAt: Map
States:
  - Type: Map
    Name: Map
    ProcessorConfig:
      ExecutionMode: Express
    ItemsPath: $Input.Items
    Processor:
      StartAt: Pass
      States:
        - Type: Pass
          Name: Pass
          End: true
    End: true

以下示例是状态机执行的输入。

{
  "Items": [
    {"key_1":"value_1"},
    {"key_2":"value_2"},
    {"key_3":"value_3"}
  ]
}

Map会产生三个子工作流执行，其中子工作流执行的定义为：

Type: StateMachine
Name: Map
SpecVersion: v1
StartAt: Pass
States:
  - Type: Pass
    Name: Pass
    End: true

以下示例显示了子工作流执行接收的输入。

# execution-1
# 第1个Map迭代对应的子工作流执行接收的输入示例
{"key_1":"value_1"}

# execution-2
# 第2个Map迭代对应的子工作流执行接收的输入示例
{"key_2":"value_2"}

# execution-3
# 第3个Map迭代对应的子工作流执行接收的输入示例
{"key_3":"value_3"}

完成后，Map状态的输出是一个JSON数组，其中每个项目都是迭代的输出。

{
    "Items": [
        {
            "key_1": "value_1"
        },
        {
            "key_2": "value_2"
        },
        {
            "key_3": "value_3"
        }
    ]
}