本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
执行多个查询（Amazon Athena、Amazon SNS）
此示例项目演示了如何连续运行 Athena 查询，然后处理错误，然后根据查询是成功还是失败发送 Amazon SNS 通知。部署此示例项目将创建Amazon Step Functions状态机、Amazon Athena 查询和 Amazon SNS 主题。
在此项目中，Step Functions 使用状态机同步运行 Athena 查询。返回查询结果后，输入并行状态，并行执行两个 Athena 查询。然后它等待任务成功或失败，并发送一个 Amazon SNS 主题，其中包含有关任务是成功还是失败的消息。
创建状态机并预置资源
-
打开Step Functions
然后选择创建状态机. -
选择Step Functions 的示例项目，然后选择执行多个查询.
此时将显示状态机 Code (代码) 和 Visual Workflow (可视工作流程)。
-
请选择 Next (下一步)。
此时将显示 Deploy resources (部署资源) 页面，其中列出了将创建的资源。对于本示例项目，资源包括：
-
Amazon Athena 查询
-
一个 Amazon SNS 主题
-
-
选择 Deploy Resources (部署资源)。
注意 创建这些资源和相关 IAM 权限可能需要长达 10 分钟的时间。当显示 Deploy resources (部署资源) 页面时，您可打开 Stack ID (堆栈 ID) 链接以查看正在预置的资源。
启动新的执行
-
在 New execution 页面上，输入执行名称 (可选)，然后选择 Start Execution (开始执行)。
（可选）要识别您的执行，您可以在名称。默认情况下，Step Functions 会自动生成唯一的执行名称。
注意 Step Functions 允许您创建包含非 ASCII 字符的状态机、执行和活动名称。这些非 ASCII 名称不适用于亚马逊CloudWatch. 为了确保你可以跟踪CloudWatch指标，请选择仅使用 ASCII 字符的名称。
-
（可选）您可以转到 Step Functions 上新创建的状态机。控制面板，然后选择新的执行.
-
执行完成后，您可以在 Visual workflow (可视工作流) 上选择状态，并浏览 Step details (步骤详细信息) 下的 Input (输入) 和 Output (输出)。
示例状态机代码
此示例项目中的状态机通过将参数直接传递给这些资源来与 Amazon Athena 和 Amazon SNS 集成。
浏览此示例状态机，Amazon Athena 解 Step Functions 如何通过连接到Resource字段，然后通过Parameters转到服务 API。
有关 Amazon Step Functions 如何控制其他 Amazon 服务的更多信息，请参阅将 Amazon Step Functions 与其他服务一起使用。
{
"Comment": "An example of using Athena to execute queries in sequence and parallel, with error handling and notifications.",
"StartAt": "Generate Example Data",
"States": {
"Generate Example Data": {
"Type": "Task",
"Resource": "arn:aws:states:::lambda:invoke",
"OutputPath": "$.Payload",
"Parameters": {
"FunctionName": "<ATHENA_FUNCTION_NAME>"
},
"Next": "Load Data to Database"
},
"Load Data to Database": {
"Type": "Task",
"Resource": "arn:aws:states:::athena:startQueryExecution.sync",
"Parameters": {
"QueryString": "<ATHENA_QUERYSTRING>",
"WorkGroup": "<ATHENA_WORKGROUP>"
},
"Catch": [
{
"ErrorEquals": [
"States.ALL"
],
"Next": "Send query results"
}
],
"Next": "Map"
},
"Map": {
"Type": "Parallel",
"ResultSelector": {
"Query1Result.$": "$[0].ResultSet.Rows",
"Query2Result.$": "$[1].ResultSet.Rows"
},
"Catch": [
{
"ErrorEquals": [
"States.ALL"
],
"Next": "Send query results"
}
],
"Branches": [
{
"StartAt": "Start Athena query 1",
"States": {
"Start Athena query 1": {
"Type": "Task",
"Resource": "arn:aws:states:::athena:startQueryExecution.sync",
"Parameters": {
"QueryString": "<ATHENA_QUERYSTRING>",
"WorkGroup": "<ATHENA_WORKGROUP>"
},
"Next": "Get Athena query 1 results"
},
"Get Athena query 1 results": {
"Type": "Task",
"Resource": "arn:aws:states:::athena:getQueryResults",
"Parameters": {
"QueryExecutionId.$": "$.QueryExecution.QueryExecutionId"
},
"End": true
}
}
},
{
"StartAt": "Start Athena query 2",
"States": {
"Start Athena query 2": {
"Type": "Task",
"Resource": "arn:aws:states:::athena:startQueryExecution.sync",
"Parameters": {
"QueryString": "<ATHENA_QUERYSTRING>",
"WorkGroup": "<ATHENA_WORKGROUP>"
},
"Next": "Get Athena query 2 results"
},
"Get Athena query 2 results": {
"Type": "Task",
"Resource": "arn:aws:states:::athena:getQueryResults",
"Parameters": {
"QueryExecutionId.$": "$.QueryExecution.QueryExecutionId"
},
"End": true
}
}
}
],
"Next": "Send query results"
},
"Send query results": {
"Type": "Task",
"Resource": "arn:aws:states:::sns:publish",
"Parameters": {
"Message.$": "$",
"TopicArn": "<SNS_TOPIC_ARN>"
},
"End": true
}
}
}IAM 示例
此示例Amazon Identity and Access Management示例项目生成的 (IAM) 策略包括执行状态机和相关资源所需的最小权限。我们建议在您的 IAM 策略仅包含这些必需的权限。
AthenaStartQueryExecution
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"athena:startQueryExecution",
"athena:stopQueryExecution",
"athena:getQueryExecution",
"athena:getDataCatalog"
],
"Resource": [
"arn:aws:athena:us-east-1:946551482283:workgroup/stepfunctions-athena-sample-project-workgroup-ztuvu9yuix",
"arn:aws:athena:us-east-1:946551482283:datacatalog/*"
]
},
{
"Effect": "Allow",
"Action": [
"s3:GetBucketLocation",
"s3:GetObject",
"s3:ListBucket",
"s3:ListBucketMultipartUploads",
"s3:ListMultipartUploadParts",
"s3:AbortMultipartUpload",
"s3:CreateBucket",
"s3:PutObject"
],
"Resource": [
"arn:aws:s3:::*"
]
},
{
"Effect": "Allow",
"Action": [
"glue:CreateDatabase",
"glue:GetDatabase",
"glue:GetDatabases",
"glue:UpdateDatabase",
"glue:DeleteDatabase",
"glue:CreateTable",
"glue:UpdateTable",
"glue:GetTable",
"glue:GetTables",
"glue:DeleteTable",
"glue:BatchDeleteTable",
"glue:BatchCreatePartition",
"glue:CreatePartition",
"glue:UpdatePartition",
"glue:GetPartition",
"glue:GetPartitions",
"glue:BatchGetPartition",
"glue:DeletePartition",
"glue:BatchDeletePartition"
],
"Resource": [
"arn:aws:glue:us-east-1:946551482283:catalog",
"arn:aws:glue:us-east-1:946551482283:database/*",
"arn:aws:glue:us-east-1:946551482283:table/*",
"arn:aws:glue:us-east-1:946551482283:userDefinedFunction/*"
]
},
{
"Effect": "Allow",
"Action": [
"lakeformation:GetDataAccess"
],
"Resource": [
"*"
]
}
]
}
AthenaGetQueryResults
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"athena:getQueryResults"
],
"Resource": [
"arn:aws:athena:us-east-1:946551482283:workgroup/*"
]
},
{
"Effect": "Allow",
"Action": [
"s3:GetObject"
],
"Resource": [
"arn:aws:s3:::*"
]
}
]
}snspPublish
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"sns:Publish"
],
"Resource": [
"arn:aws:sns:us-east-1:946551482283:StepFunctionsSample-AthenaMultipleQueriese1ec229b-5cbe-4754-a8a8-078474bac878-SNSTopic-9AID0HEJT7TH"
]
}
]
}LambdaInvoke函数
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"lambda:InvokeFunction"
],
"Resource": [
"arn:aws:lambda:us-east-1:946551482283:function:StepFunctionsSample-Athen-LambdaForStringGeneratio-GQFQjN7mE9gl:*"
]
},
{
"Effect": "Allow",
"Action": [
"lambda:InvokeFunction"
],
"Resource": [
"arn:aws:lambda:us-east-1:946551482283:function:StepFunctionsSample-Athen-LambdaForStringGeneratio-GQFQjN7mE9gl"
]
}
]
}有关在将 Step Functions 与其他配置时如何配置 IAM 的信息Amazon服务，请参阅集成服务的 IAM 策略.