実現したいこと

外部のプログラムから渡された入力のJsonを、一部追加のパラメータを加えながらそのまま各ステートに引き継ぎたい。

前提

StepFunctions, Lambdaを使用して、下の画像のような
Parallelステートを持つMapステートを作成しています。

この時、外部からCountRecordに対しては正常に入力を渡すことが出来ているのですが、
その後、Parallelステートも持つMap要素への入力がうまく渡せません。

定義の一部は下記の通りです。

json
1{
2  "StartAt": "CountRecord",
3  "States": {
4    "CountRecord": {
5      "Type": "Task",
6      "Resource": "arn:aws:lambda:ap-northeast-1:XXXXXXXXXXXX:function:lambda_count_record",
7      "InputPath": "$",
8      "ResultPath": "$.blocks.items",
9      "Next": "DynamicParallel"
10    },
11    "DynamicParallel": {
12      "Type": "Map",
13      "InputPath": "$.blocks.items",
14      "MaxConcurrency": 0,
15      "Iterator": {
16        "StartAt": "ExportCsv",
17        "States": {
18          "ExportCsv": {
19            "Type": "Parallel",
20            "End": true,
21            "Branches": [
22              {
23                "StartAt": "Thread_01",
24                "States": {
25                  "Thread_01": {
26                    "Type": "Task",
27                    "InputPath": "$",
28                    "Parameters": {
29                      "thread_num": 0
30                    },
31                    "Resource": "arn:aws:lambda:ap-northeast-1:XXXXXXXXXXXX:function:lambda_export_csv",
32                    "ResultPath": "$.output",
33                    "Next": "Thread_02"
34                  },
35                  "Thread_02": {
36                    "Type": "Task",
37                    "InputPath": "$",
38                    "Parameters": {
39                      "thread_num": 1
40                    },
41                    "Resource": "arn:aws:lambda:ap-northeast-1:XXXXXXXXXXXX:function:lambda_export_csv",
42                    "ResultPath": "$.output",
43                    "Next": "Thread_03"
44                  }
45<以下略>

具体的には、CountRecord からの output は下記のように想定通りなのですが、

json
1{
2  "aaa": "XXXXXX",
3  "bbb": [],
4  "blocks": {
5    "items": [
6      {
7        "abc": 0,
8        "def": 2002,
9      }
10    ]
11  }
12}

上記の $.blocks.items要素を Thread_01, Thread_04, Thread_07にうまく渡せておらず
ExecutionFailed となってしまっています。

エラーログは下記です。

json
1{
2  "error": "States.Runtime",
3  "cause": "Reference path \"$\" must point to array."
4}

CountRecordからの出力結果を配列としたとき、Iterator内のParallelステートの各ブランチの先頭タスクにうまくInputを渡すにはどのように修正すべきでしょうか。

ご教示いただけると幸いです。
よろしくお願いいたします。

行動規範の内容に同意します

回答1件

自己解決

自己解決しました。

具体的には、Parametersを使用した方法となりました。
各ブランチのInput要素を

json
1{
2  "StartAt": "Thread_01",
3  "States": {
4    "Thread_01": {
5       "Type": "Task",
6       "InputPath": "$",  ★
7       "Parameters": {
8         "thread_num": 0
9       },
10       "Resource": "arn:aws:lambda:ap-northeast-1:XXXXXXXXXXXX:function:lambda_export_csv",
11       "ResultPath": "$.output",
12       "Next": "Thread_02"
13    }

のような状態から

json
1{
2  "StartAt": "Thread_01",
3  "States": {
4    "Thread_01": {
5      "Type": "Task",
6      "Parameters": {
7        "params.$": "$",  ★
8        "thread_index": 1,
9        "branch_num": 0
10      },
11      "Resource": "arn:aws:lambda:ap-northeast-1:XXXXXXXXXXXX:function:lambda_export_csv",
12      "ResultPath": "$.params",
13      "Next": "Thread_02"
14    }