如何确保 Step Function 在 AWS 中正​​确执行参数化查询？

Question

我目前正在尝试在状态机期间执行 Athena 查询。查询本身需要一个日期变量以在多个 WHERE 语句中使用，因此我使用 Lambda 来生成它。

当我直接在 Athena 中运行 EXECUTE prepared-statement USING 'date', 'date', 'date'; 时，我得到了我期望的结果，所以我知道查询是正确形成的，但是当我尝试在状态机中执行它时，它给我以下错误：

SYNTAX_ERROR: line 19:37: Unexpected parameters (integer) for function date. Expected: date(varchar(x)) , date(timestamp) , date(timestamp with time zone)

所以我最好的猜测是我没有正确传递执行参数。

计算日期的 Lambda 以 %Y-%m-%d 格式的字符串返回它，在状态机中我确保将它传递给需要它的每个状态的输出。然后我得到一个命名查询来从状态机中创建一个准备语句。然后，我使用该准备好的语句运行 EXECUTE 查询，该查询多次需要日期，因此我使用一个内部函数将其转换为数组：

{
    "StartAt": "calculate_date",
    "States": {
        "calculate_date": {
            "Type": "Task",
            "Resource": "arn:aws:states:::lambda:invoke",
            "Parameters": {
                "Payload.$": "$",
                "FunctionName": "arn:aws:lambda:::function:calculate_date:$LATEST"
            },
            "Retry": [
                {
                "ErrorEquals": [
                    "Lambda.ServiceException",
                    "Lambda.AWSLambdaException",
                    "Lambda.SdkClientException",
                    "Lambda.TooManyRequestsException"
                ],
                "IntervalSeconds": 2,
                "MaxAttempts": 6,
                "BackoffRate": 2
                }
            ],
            "Next": "get_query",
            "ResultSelector": {
                "ExecDate.$": "$.Payload.body.exec_date"
            }
        },
        "get_query": {
            "Type": "Task",
            "Next": "prepare_query",
            "Parameters": {
                "NamedQueryId": "abc123"
            },
            "Resource": "arn:aws:states:::aws-sdk:athena:getNamedQuery",
            "ResultPath": "$.Payload"
        },
        "prepare_query": {
            "Type": "Task",
            "Next": "execute_query",
            "Parameters": {
                "QueryStatement.$": "$.Payload.NamedQuery.QueryString",
                "StatementName": "PreparedStatementName",
                "WorkGroup": "athena-workgroup"
            },
            "Resource": "arn:aws:states:::aws-sdk:athena:createPreparedStatement",
            "ResultPath": "$.Payload"
        },
        "execute_query": {
            "Type": "Task",
            "Resource": "arn:aws:states:::athena:startQueryExecution",
            "Parameters": {
                "ExecutionParameters.$": "States.Array($.ExecDate, $.ExecDate, $.ExecDate)",
                "QueryExecutionContext": {
                    "Catalog": "catalog_name",
                    "Database": "database_name"
                },
                "QueryString": "EXECUTE PreparedStatementName",
                "WorkGroup": "athena-workgroup",
                "ResultConfiguration": {
                    "OutputLocation": "s3://bucket"
                }
            },
            "End": true
        }
    }
}

状态机的执行成功返回，但查询不会将结果导出到存储桶，当我单击事件列表中的“Athena 查询执行”链接时，它会将我带到 Athena 编辑器页面，我在其中查看上面列出的错误

https://i.stack.imgur.com/pxxOm.png

我生成的 ExecutionParameters 错了吗？ createPreparedStatement 资源是否需要不同的查询参数语法？我真的在这里迷路了，所以非常感谢任何帮助

Answer 1

我刚刚解决了我的问题。如果有人遇到同样的问题，我会发布这个答案。

显然，Athena StartQueryExecution 状态中的 ExecutionParameters 参数不遵守 JSONPath 变量的变量类型，因此您需要在形成数组时手动添加单引号。我通过添加来自 lambda 的辅助输出解决了这个问题，日期用单引号引起来，这样当我使用内部函数创建数组并将其传递给查询执行时，它会正确地形成查询字符串。

像这样转换 lambda 的输出：

"ExecDateQuery.$": "States.Format('\'{}\'', $.Payload.body.exec_date)"

并在数组内部函数中使用ExecDateQuery，而不是ExecDate。

最终的状态机如下所示：

{
    "StartAt": "calculate_date",
    "States": {
        "calculate_date": {
            "Type": "Task",
            "Resource": "arn:aws:states:::lambda:invoke",
            "Parameters": {
                "Payload.$": "$",
                "FunctionName": "arn:aws:lambda:::function:calculate_date:$LATEST"
            },
            "Retry": [
                {
                "ErrorEquals": [
                    "Lambda.ServiceException",
                    "Lambda.AWSLambdaException",
                    "Lambda.SdkClientException",
                    "Lambda.TooManyRequestsException"
                ],
                "IntervalSeconds": 2,
                "MaxAttempts": 6,
                "BackoffRate": 2
                }
            ],
            "Next": "get_query",
            "ResultSelector": {
                "ExecDate.$": "$.Payload.body.exec_date",
                "ExecDateQuery.$": "States.Format('\'{}\'', $.Payload.body.exec_date)"
            }
        },
        "get_query": {
            "Type": "Task",
            "Next": "prepare_query",
            "Parameters": {
                "NamedQueryId": "abc123"
            },
            "Resource": "arn:aws:states:::aws-sdk:athena:getNamedQuery",
            "ResultPath": "$.Payload"
        },
        "prepare_query": {
            "Type": "Task",
            "Next": "execute_query",
            "Parameters": {
                "QueryStatement.$": "$.Payload.NamedQuery.QueryString",
                "StatementName": "PreparedStatementName",
                "WorkGroup": "athena-workgroup"
            },
            "Resource": "arn:aws:states:::aws-sdk:athena:createPreparedStatement",
            "ResultPath": "$.Payload"
        },
        "execute_query": {
            "Type": "Task",
            "Resource": "arn:aws:states:::athena:startQueryExecution",
            "Parameters": {
                "ExecutionParameters.$": "States.Array($.ExecDateQuery, $.ExecDateQuery, $.ExecDateQuery)",
                "QueryExecutionContext": {
                    "Catalog": "catalog_name",
                    "Database": "database_name"
                },
                "QueryString": "EXECUTE PreparedStatementName",
                "WorkGroup": "athena-workgroup",
                "ResultConfiguration": {
                    "OutputLocation": "s3://bucket"
                }
            },
            "End": true
        }
    }
}