如何在AWS Step Functions中传递列表至Glue作业参数

我在工作流中查询DynamoDB里load_fail状态等于1的表，若存在至少一个表，就启动Glue作业并把表列表作为--source_tables参数传入。当前状态机配置如下：

{
  "Comment": "A description of my state machine",
  "StartAt": "Query",
  "States": {
    "Query": {
      "Type": "Task",
      "Next": "Choice",
      "Parameters": {
        "TableName": "source_tables_load_status",
        "KeyConditionExpression": "load_fail = :load_fail",
        "ExpressionAttributeValues": {
          ":load_fail": {
            "S": "1"
          }
        }
      },
      "Resource": "arn:aws:states:::aws-sdk:dynamodb:query",
      "ResultSelector": {
        "count.$": "$.Count",
        "startTime.$": "$$.Execution.StartTime",
        "items.$": "$.Items[*].table_name.S"
      }
    },
    "Choice": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.count",
          "NumericGreaterThanEquals": 1,
          "Next": "start_glue"
        }
      ],
      "Default": "Success"
    },
    "start_glue": {
      "Type": "Task",
      "Resource": "arn:aws:states:::glue:startJobRun",
      "Parameters": {
        "JobName": "data-moving-glue",
        "Arguments": {
          "--dynamodb_metadata_table": "metadata_table",
          "--source_tables.$": "$.items"
        }
      },
      "End": true
    },
    "Success": {
      "Type": "Succeed"
    }
  }
}

当前配置中"--source_tables.$": "$.items"引发错误，错误内容：

An error occurred while executing the state 'start_glue' (entered at the event id #9). 
The Parameters '{"JobName":"data-moving-glue","Arguments":{"--dynamodb_metadata_table":"metadata_table","--source_tables":["dbo.Table_Two", "dbo.Table_Three"]}}' 
could not be used to start the Task: [The value for the field '--source_tables' must be a STRING]

错误原因 #

Glue作业的Arguments参数要求所有键值对的值必须是字符串类型，但目前传递的是数组，类型不匹配导致报错。

解决方案 #

方式1：将列表转为逗号分隔的字符串（推荐） #

在Step Functions的Query状态里，使用内置的States.Join函数把查询到的表名数组转为逗号分隔的字符串，传递给Glue后，在Glue作业代码中再拆分回列表。

修改后的状态机配置

{
  "Comment": "A description of my state machine",
  "StartAt": "Query",
  "States": {
    "Query": {
      "Type": "Task",
      "Next": "Choice",
      "Parameters": {
        "TableName": "source_tables_load_status",
        "KeyConditionExpression": "load_fail = :load_fail",
        "ExpressionAttributeValues": {
          ":load_fail": {
            "S": "1"
          }
        }
      },
      "Resource": "arn:aws:states:::aws-sdk:dynamodb:query",
      "ResultSelector": {
        "count.$": "$.Count",
        "startTime.$": "$$.Execution.StartTime",
        "items_str.$": "States.Join(',', $.Items[*].table_name.S)"
      }
    },
    "Choice": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.count",
          "NumericGreaterThanEquals": 1,
          "Next": "start_glue"
        }
      ],
      "Default": "Success"
    },
    "start_glue": {
      "Type": "Task",
      "Resource": "arn:aws:states:::glue:startJobRun",
      "Parameters": {
        "JobName": "data-moving-glue",
        "Arguments": {
          "--dynamodb_metadata_table": "metadata_table",
          "--source_tables.$": "$.items_str"
        }
      },
      "End": true
    },
    "Success": {
      "Type": "Succeed"
    }
  }
}

Glue作业代码处理（Python示例）

在Glue作业中接收参数并拆分：

import sys
from awsglue.utils import getResolvedOptions

# 获取传递的参数
args = getResolvedOptions(sys.argv, ['source_tables'])
# 拆分逗号分隔的字符串为列表
source_tables = args['source_tables'].split(',')

# 后续业务逻辑使用source_tables列表

方式2：使用Glue重复参数格式 #

Glue支持通过重复传递同一个参数名来传入列表（比如--source_tables dbo.Table_Two --source_tables dbo.Table_Three），但Step Functions的JSON不允许重复键，需要通过Pass状态构造参数结构，或者使用States.Array配合动态参数生成。不过这种方式相对复杂，不如逗号分隔字符串通用。

内容的提问来源于stack exchange，提问作者marcin2x4