执行多个查询(Amazon Athena、Amazon SNS) - Amazon Step Functions
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 Amazon Web Services 服务入门

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

执行多个查询(Amazon Athena、Amazon SNS)

此示例项目演示了如何连续运行 Athena 查询,然后处理错误,然后根据查询是成功还是失败发送 Amazon SNS 通知。部署此示例项目将创建Amazon Step Functions状态机、Amazon Athena 查询和 Amazon SNS 主题。

在此项目中,Step Functions 使用状态机同步运行 Athena 查询。返回查询结果后,输入并行状态,并行执行两个 Athena 查询。然后它等待任务成功或失败,并发送一个 Amazon SNS 主题,其中包含有关任务是成功还是失败的消息。

创建状态机并预置资源

  1. 打开Step Functions然后选择创建状态机.

  2. 选择Step Functions 的示例项目,然后选择执行多个查询.

    此时将显示状态机 Code (代码)Visual Workflow (可视工作流程)

    
                        运行多个查询工作流程。
  3. 请选择 Next (下一步)

    此时将显示 Deploy resources (部署资源) 页面,其中列出了将创建的资源。对于本示例项目,资源包括:

    • Amazon Athena 查询

    • 一个 Amazon SNS 主题

  4. 选择 Deploy Resources (部署资源)

    注意

    创建这些资源和相关 IAM 权限可能需要长达 10 分钟的时间。当显示 Deploy resources (部署资源) 页面时,您可打开 Stack ID (堆栈 ID) 链接以查看正在预置的资源。

启动新的执行

  1. New execution 页面上,输入执行名称 (可选),然后选择 Start Execution (开始执行)

  2. (可选)要识别您的执行,您可以在名称。默认情况下,Step Functions 会自动生成唯一的执行名称。

    注意

    Step Functions 允许您创建包含非 ASCII 字符的状态机、执行和活动名称。这些非 ASCII 名称不适用于亚马逊CloudWatch. 为了确保你可以跟踪CloudWatch指标,请选择仅使用 ASCII 字符的名称。

  3. (可选)您可以转到 Step Functions 上新创建的状态机。控制面板,然后选择新的执行.

  4. 执行完成后,您可以在 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 策略.