CloudFlow:分散モード

概念

分散モード: Object Storage Service (OSS) データが Map 状態によって処理されるモード。分散モードでは、大量のデータを同時に処理できます。OSS バケット内のオブジェクトをソースデータとして使用できます。

サブワークフロー実行: 分散モードで Map 状態を実行することは、サブワークフローを実行することと考えることができます。サブワークフローの定義は Map プロセッサの状態遷移定義であり、各サブワークフローの入力は 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

入力から配列を抽出するために使用される式。 ItemsPath 式の実行後に JSON 配列が返された場合、Map 状態を実行できます。配列内の各オブジェクトは、処理のために ItemProcessor に渡されます。次の例に示すように、$Context 変数と $Input 変数を使用して配列を抽出できます。

$Input.FieldA

Processor

Map プロセッサ。次の表は、Processor のフィールドについて説明しています。

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

ProcessorConfig

プロセッサ構成。次の表は、ProcessorConfig のフィールドについて説明しています。

ItemReader

分散モードでは、OSS バケットから読み取られるより大きなサイズの入力がサポートされています。次の表は、ItemReader のフィールドについて説明しています。

SourceParameters

データソースのパラメーター。次の表は、SourceParameters のフィールドについて説明しています。

ReaderConfig

リーダー構成。次の表は、ReaderConfig のフィールドについて説明しています。

OSS バケットからデータを読み取るには、次のいずれかの方法を使用します。

• OSS バケットに保存されている CSV ファイルからデータを読み取る

  Map 状態を使用して CSV ファイルからデータを読み取ることができます。たとえば、example-bucket という名前の OSS バケットに保存されている example-object.csv という名前の CSV ファイルがあるとします。次のサンプル スニペットを参照して、CSV ファイルからデータを読み取ることができます。

  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 配列が含まれています。

  指定された OSS バケットから JSON 配列を読み取ることができます。たとえば、example-bucket という名前の OSS バケットに保存されている example-object.json という名前の JSON ファイルがあるとします。次のサンプル スニペットを参照して、OSS バケットから 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_JSON_LIST
        SourceParameters:
          Bucket: example-bucket
          ObjectName: example-object.json
      End: true

  次のスニペットは、example-object.json ファイルの内容を示しています。

  [
    {
      "key_1": "value_1"
    }
  ]

  次のスニペットは、サブワークフロー実行のサンプル入力を示しています。

  {
    "key_1": "value_1",
  }

• OSS バケットに保存されているオブジェクトを読み取る

  指定された OSS バケットからオブジェクトを読み取ることができます。たとえば、名前に example-prefix というプレフィックスが付いたオブジェクトのリストがあるとします。オブジェクトは、example-bucket という名前の OSS バケットに保存されています。次のスニペットは、ステートマシンの定義例を示しています。

  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-bucket バケットに example-prefix/object_1 オブジェクトが保存されていることを示しています。

  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 バケットからインベントリを読み取る

  Map 状態を使用して、指定された OSS バケットまたはオブジェクトからデータを読み取ることができます。たとえば、example-bucket という名前の OSS バケットに保存されている manifest.json という名前のインベントリ ファイルがあるとします。インベントリ ファイルへのパスは、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 は、複数の項目をバッチにまとめ、サブワークフロー実行の入力として使用します。次の表は、ItemBatcher のフィールドについて説明しています。

MaxItemsPerBatch

MaxItemsPerBatch は、バッチで処理できる項目の最大数を指定します。次のスニペットは、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
# 最初のサブワークフロー実行のサンプル入力
{
  "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 は、1 つのバッチで処理できるデータの最大バイト数を指定します。次のスニペットは、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
# 最初のサブワークフロー実行のサンプル入力
{
  "Items":[
    {"Key":1},
    {"key":2}
  ]
}

# execution-2
# 2 番目のサブワークフロー実行のサンプル入力
{
  "Items":[
    {"Key":3},
    {"key":4}
  ]
}

# execution-3
# 3 番目のサブワークフロー実行のサンプル入力
{
  "Items":[
    {"Key":5}
  ]
}

BatchInput

BatchInput は、単一バッチの入力として使用されるデータセットを提供します。 BatchInput は、項目に加えて追加の入力をサポートします。次のスニペットは、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
# 最初のサブワークフロー実行のサンプル入力
{
  "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 の使用方法の例を示しています。

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
# 最初のサブワークフロー実行のサンプル入力
{
  "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 ファイルの内容を示しています。

# 次のスニペットは、ストレージ パスが example-prefix/map-run-name/manifest.json である 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
            }
        ]
    }
}

# 次のスニペットは、ストレージ パスが example-prefix/map-run-name/FAILED_0.json である 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"
    }
]

# 次のスニペットは、ストレージ パスが example-prefix/map-run-name/SUCCEED_0.json である 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 ステートマシンで実行できる項目の最大数を指定します。たとえば、OSS バケットに 10,000 個のオブジェクトが含まれているとします。 MaxItems を 1,000 に設定すると、Map ステートマシンは OSS バケットから 1,000 個のオブジェクトのみを読み込みます。

MaxConcurrency

MaxConcurrency は、同時サブワークフロー実行の数を指定します。たとえば、Map ステートマシンが 10,000 項目を実行するように構成されていて、MaxConcurrency を 100 に設定すると、Map ステートマシンは 100 個のサブワークフローを同時に実行します。

ToleratedFailurePercentage

ToleratedFailurePercentage は、操作で許容できるエラーの割合を指定します。たとえば、10,000 項目があり、ToleratedFailurePercentage を 10 に設定するとします。この場合、Map ステートマシンは最大 1,000 の失敗した項目実行を許容できます。

ToleratedFailureCount

ToleratedFailureCount は、操作で許容できるエラー回数を指定します。たとえば、10,000 項目があり、ToleratedFailureCount を 10 に設定するとします。この場合、Map ステートマシンは最大 10 の失敗した項目実行を許容できます。

例

次のワークフローは、分散モードの 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 ステートマシンは 3 つのサブワークフロー実行を生成します。次のスニペットは、サブワークフロー実行の定義を示しています。

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

次のスニペットは、サブワークフロー実行のサンプル入力を示しています。

# execution-1
# 最初の Map 状態に対応するサブワークフロー実行のサンプル入力
{"key_1":"value_1"}

# execution-2
# 2 番目の Map 状態に対応するサブワークフロー実行のサンプル入力
{"key_2":"value_2"}

# execution-3
# 3 番目の Map 状態に対応するサブワークフロー実行のサンプル入力
{"key_3":"value_3"}

前のスニペットを実行すると、Map ステートマシンの出力は JSON 配列になります。配列内の各オブジェクトは、Map 状態の出力です。

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