使用 Athena 按顺序并行执行查询

此示例项目演示了如何连续运行 Athena 查询，然后并行运行，处理错误，然后根据查询成功还是失败发送 SNS Amazon 通知。

在本项目中，Step Functions 使用状态机同步运行 Athena 查询。返回查询结果后，进入并行状态，并行执行两个 Athena 查询。然后，它会等待任务成功或失败，然后向Amazon SNS 主题发送一条关于任务成功还是失败的消息。

第 1 步：创建状态机

打开 Step Functions 控制台，然后选择创建状态机。
在搜索框中键入 Execute multiple queries，然后从返回的搜索结果中选择执行多个查询。
选择下一步以继续。
选择 “运行演示” 以创建只读和 ready-to-deploy 工作流程，或者选择 “在其上构建” 以创建可编辑的状态机定义，您可以在此基础上构建并稍后部署。

该示例项目部署了以下资源：
- Amazon Athena 查询
- 网络 ACL 和安全组都允许 (因此可到达您的实例) 的发起 ping 的 Amazon SNS topic
- 网络 ACL 和安全组都允许 (因此可到达您的实例) 的发起 ping 的 AWS Step Functions 状态机
- 相关 AWS Identity and Access Management (IAM) 角色
下图显示了执行多个查询示例项目的工作流图：
选择使用模板继续进行选择。

后续步骤取决于您之前的选择：

运行演示 — 您可以先查看状态机，然后再使用部署的资源创建只读项目 AWS CloudFormation 给你的 AWS 账户.

您可以查看状态机定义，准备就绪后，选择部署并运行以部署项目并创建资源。

部署最多可能需要 10 分钟才能创建资源和权限。您可以使用堆栈 ID 链接来监控进度 AWS CloudFormation.

部署完成后，您应该会在控制台中看到您的新状态机。
在此基础上再接再厉 — 您可以查看和编辑工作流程定义。在尝试运行自定义工作流程之前，您可能需要为示例项目中的占位符设置值。

注意

部署到您的账户的服务可能会收取标准费用。

第 2 步：运行状态机

在状态机页面上，选择您的示例项目。
在示例项目页面上，选择启动执行。
在启动执行对话框中，执行以下操作：
1. （可选）输入自定义执行名称以覆盖生成的默认值。
  
  非ASCII姓名和日志
  Step Functions 接受状态机、执行、活动和包含非ASCII字符的标签的名称。由于此类字符不适用于亚马逊 CloudWatch，因此我们建议您仅使用ASCII字符，以便您可以跟踪中的指标 CloudWatch。
2. （可选）在输入框中，将输入值输入为JSON。如果您正在运行演示，则可以跳过此步骤。
3. 选择启动执行。
Step Functions 控制台将引导您进入执行详情页面，您可以在图表视图中选择状态以浏览步骤详细信息窗格中的相关信息。

示例状态机代码

此示例项目中的状态机通过将参数直接传递给这些资源来与 Amazon Athena 和 A SNS mazon 集成。

浏览此示例状态机，了解 Step Functions 如何通过连接到字段Resource中的SNS亚马逊资源ARN名称 () 并传递到服务Parameters来控制亚马逊 Athena 和亚马逊。API

有关如何做的更多信息 AWS Step Functions 可以控制其他 AWS 服务，请参阅将服务与 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示例

此示例 AWS 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-2:123456789012:workgroup/stepfunctions-athena-sample-project-workgroup-ztuvu9yuix",
                "arn:aws:athena:us-east-2:123456789012: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-2:123456789012:catalog",
                "arn:aws:glue:us-east-2:123456789012:database/*",
                "arn:aws:glue:us-east-2:123456789012:table/*",
                "arn:aws:glue:us-east-2:123456789012:userDefinedFunction/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "lakeformation:GetDataAccess"
            ],
            "Resource": [
                "*"
            ]
        }
    ]
}

AthenaGetQueryResults


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "athena:getQueryResults"
            ],
            "Resource": [
                "arn:aws:us-east-2:123456789012:workgroup/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject"
            ],
            "Resource": [
                "arn:aws:s3:::*"
            ]
        }
    ]
}

SNSPublish


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "sns:Publish"
            ],
            "Resource": [
                "arn:aws:sns:us-east-2:123456789012:StepFunctionsSample-AthenaMultipleQueriese1ec229b-5cbe-4754-a8a8-078474bac878-SNSTopic-9AID0HEJT7TH"
            ]
        }
    ]
}

LambdaInvokeFunction


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction"
            ],
            "Resource": [
                "arn:aws:lambda:us-east-2:123456789012:function:StepFunctionsSample-Athen-LambdaForStringGeneratio-GQFQjN7mE9gl:*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction"
            ],
            "Resource": [
                "arn:aws:lambda:us-east-2:123456789012:function:StepFunctionsSample-Athen-LambdaForStringGeneratio-GQFQjN7mE9gl"
            ]
        }
    ]
}

有关在将 Step Functions 与其他功能一起使用IAM时如何进行配置的信息 AWS 服务，请参阅Step Functions 如何为集成服务生成IAM策略。

Javascript 在您的浏览器中被禁用或不可用。

要使用 Amazon Web Services 文档，必须启用 Javascript。请参阅浏览器的帮助页面以了解相关说明。

文档惯例

启动 Athena 查询

查询大型数据集