本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 扇出到 Firehose 传输流 您可以为 Amazon SNS 主题订阅[传输流](https://docs.aws.amazon.com/firehose/latest/dev/what-is-this-service.html)，从而使您能够将通知发送到其他存储和分析端点。发布到 Amazon SNS 主题的消息将发送到订阅的 Firehose 传输流，并按照 Firehose 中配置的方式传输到目标。订阅所有者最多可以为 Amazon SNS 主题订阅五个 Firehose 传输流。每个 Firehose 传输流对于每秒请求量和吞吐量都有一个[原定设置限额](https://docs.aws.amazon.com/firehose/latest/dev/limits.html)。此限制可能会导致发布的消息（入站流量）多于传输的消息（出站流量）。当入站流量多于出站流量时，您的订阅可能会累积大量的消息积压，从而可能会导致较长的消息传输延迟。您可以根据发布率请求[增加限额](https://support.console.aws.amazon.com/support/home#/case/create?issueType=service-limit-increase)，以避免对您的工作负载产生不利影响。 通过Firehose传送流，您可以将亚马逊SNS通知分发给亚马逊简单存储服务（Amazon S3）、亚马逊Redshift、亚马逊服务（服务）以及第三方OpenSearch 服务提供商，例如Datadog、 OpenSearch New Relic、MongoDB和Splunk。 例如，您可以使用此功能将发送到 Amazon S3 存储桶中的主题的消息永久存储以用于合规性、存档或其他目的。为此，请使用 Amazon S3 存储桶目标创建 Firehose 传输流，并为 Amazon SNS 主题订阅该传输流。再举一个例子，要对发送到 Amazon SNS 主题的消息进行分析，请创建带有 OpenSearch 服务索引目标的传输流。然后，您可以为 Amazon SNS 主题订阅 Firehose 传输流。 Amazon SNS 还支持针对发送到 Firehose 端点的通知进行消息传输状态日志记录。有关更多信息，请参阅 [Amazon SNS 消息传输状态](sns-topic-attributes.md)。 # 订阅 Firehose 传输流到 Amazon SNS 主题的先决条件 要订阅 SNS 主题的直播流，您 AWS 账户 必须具备以下条件： + 标准 SNS 主题。有关更多信息，请参阅 [创建 Amazon SNS 主题](sns-create-topic.md)。 + Firehose 传输流。有关更多信息，请参阅《Amazon Data Firehose 开发人员指南》**中的[创建传输流](https://docs.aws.amazon.com/firehose/latest/dev/basic-create.html)和[授予应用程序对 Firehose 资源的访问权限](https://docs.aws.amazon.com/firehose/latest/dev/controlling-access.html#access-to-firehose)。 + 信任 Amazon SNS 服务委托人并有权写入传输流的 AWS Identity and Access Management (IAM) 角色。创建订阅时，您将输入此角色的 Amazon Resource Name (ARN) 作为 `SubscriptionRoleARN`。Amazon SNS 担任此角色，它允许 Amazon SNS 将记录放入 Firehose 传输流中。 下面的示例策略显示了建议的权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Action": [ "firehose:DescribeDeliveryStream", "firehose:ListDeliveryStreams", "firehose:ListTagsForDeliveryStream", "firehose:PutRecord", "firehose:PutRecordBatch" ], "Resource": [ "arn:aws:firehose:us-east-1:111111111111:deliverystream/firehose-sns-delivery-stream" ], "Effect": "Allow" } ] } ``` ------ 要提供使用 Firehose 的完全权限，您还可以使用 AWS 托管策略。`AmazonKinesisFirehoseFullAccess`或者，要为使用 Firehose 提供更严格的权限，您可以创建自己的策略。至少，策略必须提供在特定传输流上运行 `PutRecord` 操作的权限。 在所有情况下，您都还必须编辑信任关系以包括 Amazon SNS 服务委托人。例如： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "sns.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } ``` ------ 有关创建角色的更多信息，请参阅 *IAM 用户指南*中的[创建角色以向 AWS 服务委派权限](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-service.html)。 完成这些要求后，您可以[将传输流订阅到 SNS 主题](firehose-endpoints-subscribe.md)。 # 将 Firehose 传输流订阅到 Amazon SNS 主题 要将 Amazon SNS 通知传输至[传输流](sns-firehose-as-subscriber.md)，首先确保您已符合所有[先决条件](prereqs-kinesis-data-firehose.md)。有关受支持端点的列表，请参阅《Amazon Web Services 一般参考》**中的[端点和配额](https://docs.aws.amazon.com/general/latest/gr/fh.html)。 **将 Firehose 传输流订阅到主题** 1. 登录 [Amazon SNS 控制台](https://console.aws.amazon.com/sns/home)。 1. 在导航窗格中，选择**订阅**。 1. 在**订阅**页面上，选择**创建订阅**。 1. 在**创建订阅**页上的**详细信息**部分，执行以下操作： 1. 对于 **Topic ARN**（主题 ARN），选择标准主题的 Amazon Resource Name (ARN)。 1. 对于**协议**，选择 **Firehose**。 1. 对于**端点**，选择可从 Amazon SNS 中接收通知的 Firehose 传输流的 ARN。 1. 对于**订阅角色 ARN**，指定您为写入到 Firehose 传输流创建的 AWS Identity and Access Management （IAM）角色的 ARN。有关更多信息，请参阅 [订阅 Firehose 传输流到 Amazon SNS 主题的先决条件](prereqs-kinesis-data-firehose.md)。 1. （可选）要从已发布消息中删除任何 Amazon SNS 元数据，请选择 **Enable raw message delivery**（启用原始消息传输）。有关更多信息，请参阅 [Amazon SNS 原始消息传输](sns-large-payload-raw-message-delivery.md)。 1. （可选）要配置筛选策略，请展开 **Subscription filter policy**（订阅筛选策略）部分。有关更多信息，请参阅 [Amazon SNS 订阅筛选策略](sns-subscription-filter-policies.md)。 1. （可选）要为订阅配置死信队列，请展开 **Redrive policy (dead-letter queue)**（重新驱动策略（死信队列））部分。有关更多信息，请参阅 [Amazon SNS 死信队列](sns-dead-letter-queues.md)。 1. 选择**创建订阅**。 控制台将创建订阅并打开订阅的 **Details**（详细信息）页面。 # 管理跨多个传送流目标的 Amazon SNS 消息 [传输流](sns-firehose-as-subscriber.md)允许您管理跨多个目的地的 Amazon SNS 消息，从而实现与亚马逊 S3、亚马逊 OpenSearch 服务、Amazon Redshift 和 HTTP 终端节点的集成，用于存储、索引和分析。通过正确配置消息格式和传送，您可以将 Amazon SNS 通知存储在 Amazon S3 中以供日后处理，使用 Amazon Athena 分析结构化消息数据，为消息编制索引以进行实时搜索和可视化， OpenSearch 在 Amazon Redshift 中结构化存档以进行高级查询。 # 分析 Amazon SNS 消息并将其存储在 Amazon S3 目标中 本主题介绍传输流如何将数据发布到 Amazon Simple Storage Service（Amazon S3）。 ![\[Amazon 消息处理服务的集成和工作流程。它显示了发布者如何向 Amazon SNS 主题发送消息，然后该主题将消息扇出到多个 Amazon SQS 队列和 Data Firehose 传输流。然后，可通过 Lambda 函数来处理消息，也可以永久存储在 Amazon S3 存储桶中。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/firehose-architecture-s3.png) **Topics** + [格式化通知以存储在 Amazon S3 目标中](firehose-archived-message-format-S3.md) + [使用 Athena 分析在 Amazon S3 中存储的消息](firehose-message-analysis-s3.md) # 格式化 Amazon SNS 通知以存储在 Amazon S3 目标中 以下示例显示了一条发送到 Amazon Simple Storage Service (Amazon S3) 存储桶的 Amazon SNS 通知，其中使用了缩进以增强可读性。 **注意** 在此示例中，已发布消息的原始消息传输被禁用。禁用原始邮件传输时，Amazon SNS 会将 JSON 元数据添加到消息中，其中包括以下属性： `Type` `MessageId` `TopicArn` `Subject` `Timestamp` `UnsubscribeURL` `MessageAttributes` 有关原始消息传输的更多信息，请参阅 [Amazon SNS 原始消息传输](sns-large-payload-raw-message-delivery.md)。 ``` { "Type": "Notification", "MessageId": "719a6bbf-f51b-5320-920f-3385b5e9aa56", "TopicArn": "arn:aws:sns:us-east-1:333333333333:my-kinesis-test-topic", "Subject": "My 1st subject", "Message": "My 1st body", "Timestamp": "2020-11-26T23:48:02.032Z", "UnsubscribeURL": "https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-east-1:333333333333:my-kinesis-test-topic:0b410f3c-ee5e-49d8-b59b-3b4aa6d8fcf5", "MessageAttributes": { "myKey1": { "Type": "String", "Value": "myValue1" }, "myKey2": { "Type": "String", "Value": "myValue2" } } } ``` 以下示例显示了通过一个传输流发送到同一 Amazon S3 存储桶的三条 SNS 消息。消息进行了缓冲，每条消息之间以换行分隔。 ``` {"Type":"Notification","MessageId":"d7d2513e-6126-5d77-bbe2-09042bd0a03a","TopicArn":"arn:aws:sns:us-east-1:333333333333:my-kinesis-test-topic","Subject":"My 1st subject","Message":"My 1st body","Timestamp":"2020-11-27T00:30:46.100Z","UnsubscribeURL":"https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-east-1:313276652360:my-kinesis-test-topic:0b410f3c-ee5e-49d8-b59b-3b4aa6d8fcf5","MessageAttributes":{"myKey1":{"Type":"String","Value":"myValue1"},"myKey2":{"Type":"String","Value":"myValue2"}}} {"Type":"Notification","MessageId":"0c0696ab-7733-5bfb-b6db-ce913c294d56","TopicArn":"arn:aws:sns:us-east-1:333333333333:my-kinesis-test-topic","Subject":"My 2nd subject","Message":"My 2nd body","Timestamp":"2020-11-27T00:31:22.151Z","UnsubscribeURL":"https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-east-1:313276652360:my-kinesis-test-topic:0b410f3c-ee5e-49d8-b59b-3b4aa6d8fcf5","MessageAttributes":{"myKey1":{"Type":"String","Value":"myValue1"}}} {"Type":"Notification","MessageId":"816cd54d-8cfa-58ad-91c9-8d77c7d173aa","TopicArn":"arn:aws:sns:us-east-1:333333333333:my-kinesis-test-topic","Subject":"My 3rd subject","Message":"My 3rd body","Timestamp":"2020-11-27T00:31:39.755Z","UnsubscribeURL":"https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-east-1:313276652360:my-kinesis-test-topic:0b410f3c-ee5e-49d8-b59b-3b4aa6d8fcf5"} ``` # 使用 Athena 分析在 Amazon S3 中存储的 Amazon SNS 消息 本页面介绍如何分析通过传输流发送到 Amazon Simple Storage Service（Amazon S3）目标的 Amazon SNS 消息。 **要分析通过 Firehose 传输流发送到 Amazon S3 目标的 SNS 消息** 1. 配置 Amazon S3 资源。有关说明，请参阅 *Amazon Simple Storage Service 用户指南*中的[创建存储桶](https://docs.aws.amazon.com/AmazonS3/latest/userguide/CreatingABucket.html)和 *Amazon Simple Storage Service 用户指南*中的[使用 Amazon S3 存储桶](https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingBucket.html)。 1. 配置传输流。有关说明，请参阅《Amazon Data Firehose 开发人员指南》**中的[选择 Amazon S3 作为目标](https://docs.aws.amazon.com/firehose/latest/dev/create-destination.html#create-destination-s3)。 1. 使用 [Amazon Athena](https://console.aws.amazon.com/athena) 通过标准的 SQL 查询 Amazon S3 对象。有关更多信息，请参阅 *Amazon Athena 用户指南*中的[入门](https://docs.aws.amazon.com/athena/latest/ug/getting-started.html)。 ## 示例查询 在本示例查询中，我们假设满足以下条件： + 消息存储在 `default` schema 的 `notifications` 表中。 + `notifications` 表包含一个类型为 `string` 的 `timestamp` 列。 以下查询返回在指定日期范围内收到的所有 SNS 消息： ``` SELECT * FROM default.notifications WHERE from_iso8601_timestamp(timestamp) BETWEEN TIMESTAMP '2020-12-01 00:00:00' AND TIMESTAMP '2020-12-02 00:00:00'; ``` # 将 Amazon SNS 消息与亚马逊 OpenSearch 服务目标集成 本节介绍交付流如何将数据发布到亚马逊 OpenSearch 服务（OpenSearch 服务）。 ![\[发布者向 Amazon SNS 主题发送消息，该主题随后将这些消息分发到多个 Amazon SQS 队列。来自这些队列的消息可以由 Lambda 函数处理，也可以通过 Data Firehose 传输流发送到亚马逊 OpenSearch 服务，从而创建可搜索的消息索引。此设置演示了使用 AWS 服务的高级消息路由和处理方案。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/firehose-architecture-es.png) **Topics** + [存档消息格式](firehose-archived-message-format-elasticsearch.md) + [分析消息](firehose-message-analysis-elasticsearch.md) # 在服务索引中 OpenSearch 存储和格式化 Amazon SNS 通知 以下示例演示了向名为的亚马逊 OpenSearch 服务（OpenSearch 服务）索引发送的 Amazon SNS 通知。`my-index`此索引在 `Timestamp` 字段中具有时间筛选字段。SNS 通知放置在负载的 `_source` 属性中。 **注意** 在此示例中，已发布消息的原始消息传输被禁用。禁用原始邮件传输时，Amazon SNS 会将 JSON 元数据添加到消息中，其中包括以下属性： `Type` `MessageId` `TopicArn` `Subject` `Timestamp` `UnsubscribeURL` `MessageAttributes` 有关原始消息传输的更多信息，请参阅 [Amazon SNS 原始消息传输](sns-large-payload-raw-message-delivery.md)。 ``` { "_index": "my-index", "_type": "_doc", "_id": "49613100963111323203250405402193283794773886550985932802.0", "_version": 1, "_score": null, "_source": { "Type": "Notification", "MessageId": "bf32e294-46e3-5dd5-a6b3-bad65162e136", "TopicArn": "arn:aws:sns:us-east-1:111111111111:my-topic", "Subject": "Sample subject", "Message": "Sample message", "Timestamp": "2020-12-02T22:29:21.189Z", "UnsubscribeURL": "https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-east-1:111111111111:my-topic:b5aa9bc1-9c3d-452b-b402-aca2cefc63c9", "MessageAttributes": { "my_attribute": { "Type": "String", "Value": "my_value" } } }, "fields": { "Timestamp": [ "2020-12-02T22:29:21.189Z" ] }, "sort": [ 1606948161189 ] } ``` # 分析服务目标的 Amazon SNS OpenSearch 消息 本主题介绍如何分析通过传送流发送到亚马逊 OpenSearch 服务（OpenSearch 服务）目标的 Amazon SNS 消息。 **分析通过 Firehose 发送到服务目标的传送流 OpenSearch 的 SNS 消息** 1. 配置您的 OpenSearch 服务资源。有关说明，请参阅《[亚马逊 OpenSearch 服务*开发者指南》中的 “亚马逊 OpenSearch 服务*入门](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/es-gsg.html)”。 1. 配置传输流。有关说明，请参阅 *Amazon Data Firehose 开发者*指南中的[为您的目的地选择 OpenSearch 服务](https://docs.aws.amazon.com/firehose/latest/dev/create-destination.html#create-destination-elasticsearch)。 1. 使用 OpenSearch 服务查询和 Kibana 运行查询。有关更多信息，请参阅《*亚马逊 OpenSearch 服务开发者*指南》[中的步骤 3：在 OpenSearch 服务域中搜索文档](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/es-gsg-search.html)和 [Kibana](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/es-kibana.html)。 ## 示例查询 以下示例查询指定日期范围内接收的所有 SNS 消息的 `my-index` 索引： ``` POST https://search-my-domain.us-east-1.es.amazonaws.com/my-index/_search { "query": { "bool": { "filter": [ { "range": { "Timestamp": { "gte": "2020-12-08T00:00:00.000Z", "lte": "2020-12-09T00:00:00.000Z", "format": "strict_date_optional_time" } } } ] } } } ``` # 在 Amazon Redshift 目标中配置 Amazon SNS 消息传输和分析 本主题介绍如何将 Amazon SNS 通知扇出到传输流，然后传输流将数据发布到 Amazon Redshift。通过此设置，您可以连接到 Amazon Redshift 数据库，并使用 SQL 查询工具检索符合特定标准的 Amazon SNS 消息。 ![\[发送者向 Amazon SNS 主题发布的消息将分发到多个 Amazon SQS 队列由 Lambda 函数进行处理，还会通过 Data Firehose 传输流发送到 Amazon Redshift 集群，以便在消息数据仓库中存储和分析。此设置使用 AWS 服务演示了强大的消息处理和数据仓库架构。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/firehose-architecture-rs.png) **Topics** + [在 Amazon Redshift 表中构造消息归档](firehose-archive-table-structure-redshift.md) + [分析存储在 Amazon Redshift 目标中的消息](firehose-message-analysis-redshift.md) # 在 Amazon Redshift 表中构造 Amazon SNS 消息归档 对于 Amazon Redshift 端点，Amazon SNS 消息会作为表中的行进行归档。下面是数据存储方式的示例： **注意** 在此示例中，已发布消息的原始消息传输被禁用。禁用原始邮件传输时，Amazon SNS 会将 JSON 元数据添加到消息中，其中包括以下属性： `Type` `MessageId` `TopicArn` `Subject` `Message` `Timestamp` `UnsubscribeURL` `MessageAttributes` 有关原始消息传输的更多信息，请参阅 [Amazon SNS 原始消息传输](sns-large-payload-raw-message-delivery.md)。 尽管 Amazon SNS 使用此列表中显示的大写向邮件添加了属性，但 Amazon Redshift 表中的列名称以所有小写字符显示。要转换 Amazon Redshift 终端节点的 JSON 元数据，您可以使用 SQL `COPY` 命令。有关更多信息，请参阅 *Amazon Redshift 数据库开发人员指南*中的[从 JSON 中复制示例](https://docs.aws.amazon.com/redshift/latest/dg/r_COPY_command_examples.html#r_COPY_command_examples-copy-from-json)和[使用“auto ignorecase”选项从 JSON 数据中加载](https://docs.aws.amazon.com/redshift/latest/dg/r_COPY_command_examples.html#copy-from-json-examples-using-auto-ignorecase)。 | 类型 | messageid | topicarn | subject | message | timestamp | unsubscribeurl | messageattributes | | --- | --- | --- | --- | --- | --- | --- | --- | | 通知 | ea544832-a0d8-581d-9275-108243c46103 | arn:aws:sns:us-east-1:111111111111:my-topic | 示例主题 | 示例消息 | 2020-12-02T00:33:32.272Z | https://sns.us-east-1.amazonaws.com/？ Action=unsubscribe& =arnSubscriptionArn: aws: sns: us-east-1:111111111111:my-topic: 326deeeb-cbf4-45da-b92b-ca77a247813b | \$1\$1"my\$1attribute\$1":\$1\$1"Type\$1":\$1"String\$1",\$1"Value\$1":\$1"my\$1value\$1"\$1\$1 | | 通知 | ab124832-a0d8-581d-9275-108243c46114 | arn:aws:sns:us-east-1:111111111111:my-topic | 示例主题 2 | 示例消息 2 | 2020-12-03T00:18:11.129Z | https://sns.us-east-1.amazonaws.com/？ Action=unsubscribe& =arnSubscriptionArn: aws: sns: us-east-1:111111111111:my-topic: 326deeeb-cbf4-45da-b92b-ca77a247813b | \$1\$1"my\$1attribute2\$1":\$1\$1"Type\$1":\$1"String\$1",\$1"Value\$1":\$1"my\$1value\$1"\$1\$1 | | 通知 | ce644832-a0d8-581d-9275-108243c46125 | arn:aws:sns:us-east-1:111111111111:my-topic | 示例主题 3 | 示例消息 3 | 2020-12-09T00:08:44.405Z | https://sns.us-east-1.amazonaws.com/？ Action=unsubscribe& =arnSubscriptionArn: aws: sns: us-east-1:111111111111:my-topic: 326deeeb-cbf4-45da-b92b-ca77a247813b | \$1\$1"my\$1attribute3\$1":\$1\$1"Type\$1":\$1"String\$1",\$1"Value\$1":\$1"my\$1value\$1"\$1\$1 | 有关向 Amazon Redshift 终端节点扇出通知的更多信息，请参阅 [在 Amazon Redshift 目标中配置 Amazon SNS 消息传输和分析](firehose-redshift-destinations.md)。 # 分析存储在 Amazon Redshift 目标中的 Amazon SNS 消息 本主题介绍如何分析通过传输流发送到 Amazon Redshift 目标的 Amazon SNS 消息。 **要分析通过 Firehose 传输流发送到 Amazon Redshift 目标的 SNS 消息** 1. 配置您的 Amazon Redshift 资源。有关说明，请参阅 *Amazon Redshift 入门指南*中的[Amazon Redshift 入门](https://docs.aws.amazon.com/redshift/latest/gsg/getting-started.html)。 1. 配置传输流。有关说明，请参阅《Amazon Data Firehose 开发人员指南》**中的[选择 Amazon Redshift 作为目标](https://docs.aws.amazon.com/firehose/latest/dev/create-destination.html#create-destination-redshift)。 1. 运行查询。有关更多信息，请参阅《Amazon Redshift 管理指南》**中的 [使用查询编辑器查询数据库](https://docs.aws.amazon.com/redshift/latest/mgmt/query-editor.html)。 ## 示例查询 在本示例查询中，我们假设满足以下条件： + 消息存储在默认 `public` schema 的 `notifications` 表中。 + SNS 消息的 `Timestamp` 属性存储在表的 `timestamp` 列中，其列数据类型为 `timestamptz`。 **注意** 要转换 Amazon Redshift 终端节点的 JSON 元数据，您可以使用 SQL `COPY` 命令。有关更多信息，请参阅 *Amazon Redshift 数据库开发人员指南*中的[从 JSON 中复制示例](https://docs.aws.amazon.com/redshift/latest/dg/r_COPY_command_examples.html#r_COPY_command_examples-copy-from-json)和[使用“auto ignorecase”选项从 JSON 数据中加载](https://docs.aws.amazon.com/redshift/latest/dg/r_COPY_command_examples.html#copy-from-json-examples-using-auto-ignorecase)。 以下查询返回在指定日期范围内收到的所有 SNS 消息： ``` SELECT * FROM public.notifications WHERE timestamp > '2020-12-01T09:00:00.000Z' AND timestamp < '2020-12-02T09:00:00.000Z'; ``` # 配置到 HTTP 目标的 Amazon SNS 消息传输 本主题介绍传输流如何将数据发布到 HTTP 端点。 ![\[Amazon SNS 主题的发布者，该主题随后会将这些消息分发到多个 Amazon SQS 队列。这些消息由 Lambda 函数处理，也通过 Data Firehose 传输流发送到 HTTP 端点。此设置展示了 AWS 服务 如何协同工作以促进消息处理以及与外部 HTTP 服务的集成。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/firehose-architecture-http.png) **Topics** + [传送到 HTTP 目标的通知格式](firehose-delivered-message-format-http.md) # 传送至 HTTP 目标的 Amazon SNS 通知格式 以下是通过传输流发送到 HTTP 端点的 Amazon SNS 的 HTTP POST 请求正文示例。Amazon SNS 通知被编码为 base64 格式的有效载荷，存放在 records 属性中。 **注意** 在此示例中，已发布消息的原始消息传输被禁用。有关原始消息传输的更多信息，请参阅 [Amazon SNS 原始消息传输](sns-large-payload-raw-message-delivery.md)。 ``` "body": { "requestId": "ebc9e8b2-fce3-4aef-a8f1-71698bf8175f", "timestamp": 1606255960435, "records": [ { "data": "eyJUeXBlIjoiTm90aWZpY2F0aW9uIiwiTWVzc2FnZUlkIjoiMjFkMmUzOGQtMmNhYi01ZjYxLTliYTItYmJiYWFhYzg0MGY2IiwiVG9waWNBcm4iOiJhcm46YXdzOnNuczp1cy1lYXN0LTE6MTExMTExMTExMTExOm15LXRvcGljIiwiTWVzc2FnZSI6IlNhbXBsZSBtZXNzYWdlIGZvciBBbWF6b24gS2luZXNpcyBEYXRhIEZpcmVob3NlIGVuZHBvaW50cyIsIlRpbWVzdGFtcCI6IjIwMjAtMTEtMjRUMjI6MDc6MzEuNjY3WiIsIlVuc3Vic2NyaWJlVVJMIjoiaHR0cHM6Ly9zbnMudXMtZWFzdC0xLmFtYXpvbmF3cy5jb20vP0FjdGlvbj1VbnN1YnNjcmliZSZTdWJzY3JpcHRpb25Bcm49YXJuOmF3czpzbnM6MTExMTExMTExMTExOm15LXRvcGljOjAxYjY5MTJjLTAwNzAtNGQ4Yi04YjEzLTU1NWJmYjc2ZTdkNCJ9" } ] } ``` # Amazon SNS 消息归档和分析：航空票务平台示例应用场景 本主题提供用于归档和分析 Amazon SNS 消息的常见使用案例教程。 此使用案例的设置是一个在受监管环境中运行的航空公司票务平台。 1. 该平台受合规性框架的约束，要求公司将所有售票记录存档至少五年。 1. 为了实现数据留存方面的合规性目标，该公司将传输流订阅到现有的 SNS 主题。 1. 传输流的目标是 Amazon Simple Storage Service (Amazon S3) 存储桶。通过此配置，发布到 SNS 主题的所有事件都将存档到 Amazon S3 存储桶中。 下图显示了此配置的架构： ![\[航空公司票务平台的 AWS 架构，说明了机票销售数据的处理和存档方式。它显示了从 Lambda 函数流经 Amazon SNS 主题的数据流，然后该主题将消息分发到 Amazon SQS 队列进行付款处理和欺诈检测，由相应的 Lambda 函数处理。数据还通过 Data Firehose 流式传输到 Amazon S3 存储桶进行长期归档，从而支持遵守数据留存要求。此设置使平台能够使用诸如 Amazon Athena 之类的工具对门票销售数据进行详细分析。\]](http://docs.aws.amazon.com/zh_cn/sns/latest/dg/images/sns-archiving-use-case.png) 为了运行分析并了解门票销售的情况，该公司使用 Amazon Athena 运行 SQL 查询。例如，公司可以通过查询来了解最受欢迎的目的地和最频繁的旅客。 要为此用例创建 AWS 资源，您可以使用 AWS 管理控制台 或 CloudFormation 模板。 **Topics** + [设置用于邮件存档和分析的初始 AWS 资源](firehose-example-initial-resources.md) + [设置用于消息归档的 Firehose 传输流](firehose-example-create-delivery-stream.md) + [将传输流订阅到主题](firehose-example-subscribe-delivery-stream-to-topic.md) + [测试和查询配置以实现有效的数据管理](firehose-example-test-and-query.md) + [使用模板自动归档邮件 CloudFormation](firehose-example-cfn.md) # 为 Amazon SNS 消息存档和分析设置初始 AWS 资源 本主题介绍如何为[消息归档和分析示例使用案例](firehose-example-use-case.md)创建所需资源： + 一个 Amazon Simple Storage Service (Amazon S3) 存储桶 + 两个 Amazon Simple Queue Service (Amazon SQS) 队列 + 一个 Amazon SNS 主题 + 对 Amazon SNS 主题的两个 Amazon SQS 订阅 **要创建起始资源** 1. 创建 Amazon S3 存储桶： 1. 打开 [Amazon S3 控制台](https://console.aws.amazon.com/s3/home)。 1. 选择**创建存储桶**。 1. 对于 **Bucket name**（存储桶名称），请输入全局唯一名称。保留其他字段作为默认值。 1. 选择 **创建存储桶 **。 有关 Amazon S3 存储桶的更多信息，请参阅 *Amazon Simple Storage Service 用户指南*中的[创建存储桶](https://docs.aws.amazon.com/AmazonS3/latest/userguide/CreatingABucket.html)和 *Amazon Simple Storage Service 用户指南*中的[使用 Amazon S3 存储桶](https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingBucket.html)。 1. 创建两个 Amazon SQS 队列： 1. 打开 [Amazon SQS 控制台](https://console.aws.amazon.com/sqs/home)。 1. 选择**创建队列**。 1. 对于 **Type**（类型），选择 **Standard**（标准）。 1. 对于**名称**，请输入 **ticketPaymentQueue**。 1. 在 **Access policy**（访问策略）下，对于 **Choose method**（选择方法），选择 **Advanced**（高级）。 1. 在 JSON 策略框中，粘贴以下策略： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "sns.amazonaws.com" }, "Action": "sqs:SendMessage", "Resource": "*", "Condition": { "ArnEquals": { "aws:SourceArn": "arn:aws:sns:us-east-1:123456789012:ticketTopic" } } } ] } ``` ------ 在此访问策略中，将 AWS 账户 数字 (*123456789012*) 替换为您自己的数字，然后相应地更改 AWS 区域 (*us-east-1*)。 1. 选择**创建队列**。 1. 重复这些步骤以创建第二个名为 **ticketFraudQueue** 的 SQS 队列。 有关创建 SQS 队列的更多信息，请参阅 *Amazon Simple Queue Service 开发人员指南*中的[创建 Amazon SQS 队列（控制台）](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-configure-create-queue.html)。 1. 创建 SNS 主题： 1. 打开 Amazon SNS 控制台中的[主题](https://console.aws.amazon.com/sns/home#/topics)页面。 1. 选择**创建主题**。 1. 在 **Details**（详细信息）下，对于 **Type**（类型），选择 **Standard**（标准）。 1. 对于**名称**，请输入 **ticketTopic**。 1. 选择**创建主题**。 有关创建 SNS 主题的更多信息，请参阅 [创建 Amazon SNS 主题](sns-create-topic.md)。 1. 同时为两个 SQS 队列订阅 SNS 主题。 1. 在 [Amazon SNS 控制台](https://console.aws.amazon.com/sns/home#/topics)上的在 **ticketTopic** 主题的详细信息页面，选择 **Create subscription**（创建订阅）。 1. 在 **Details**（详细信息）下，对于 **Protocol**（方案），选择 **Amazon SQS**。 1. 对于**终端节点**，选择队列的亚马逊资源名称 (ARN)。**ticketPaymentQueue** 1. 选择**创建订阅**。 1. 重复这些步骤，使用队列的 ARN 创建第二个订阅。**ticketFraudQueue** 有关订阅 SNS 主题的更多信息，请参阅 [在 Amazon SNS 主题中创建订阅](sns-create-subscribe-endpoint-to-topic.md)。您还可以从 Amazon SQS 控制台为 SQS 队列订阅 SNS 主题。有关更多信息，请参阅 *Amazon Simple Queue Service 开发人员指南*中的[将 Amazon SQS 队列订阅到 Amazon SNS 主题（控制台）](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-configure-subscribe-queue-sns-topic.html)。 您已为此示例使用案例创建了初始资源。要继续，请参阅 [为 Amazon SNS 消息归档设置 Amazon Data Firehose 传输流](firehose-example-create-delivery-stream.md)。 # 为 Amazon SNS 消息归档设置 Amazon Data Firehose 传输流 本主题介绍如何为[消息归档和分析示例使用案例](firehose-example-use-case.md)创建 Amazon Data Firehose 传输流。 **创建 Amazon Data Firehose 传输流** 1. 打开 [Amazon Data Firehose 服务控制台](https://console.aws.amazon.com/kinesis/home)。 1. 选择 **Firehose**，然后选择**创建传输流**。 1. 在 **New delivery stream**（新传输流）页面上，对于 **Delivery stream name**（传输流名称），输入 **ticketUploadStream**，然后选择 **Next**（下一步）。 1. 在 **Process records**（处理记录）页面上，选择 **Next**（下一步）。 1. 在 **Choose a destination**（选择目标）页面上，执行以下操作： 1. 对于 **Destination**（目标），选择 **Amazon S3**。 1. 在 **S3 destination**（S3 目标）下，对于 **S3 bucket**（S3 存储桶），选择您[最初创建的](firehose-example-initial-resources.md) S3 存储桶。 1. 选择**下一步**。 1. 在 **Configure settings**（配置设置）页面上，对于 **S3 缓冲区条件**，执行以下操作： + 对于 **Buffer size**（缓冲区大小），输入 **1**。 + 对于 **Buffer interval**（缓冲区时间间隔），输入 **60**。 通过将这些值用于 Amazon S3 缓冲区，您可以快速测试配置。满足的第一个缓冲条件将触发至 S3 存储桶的数据传输。 1. 在**配置设置**页面上，对于**权限**，选择创建自动分配所需权限的 AWS Identity and Access Management (IAM) 角色。然后选择**下一步**。 1. 在 **Review**（审核）页面上，选择 **Create delivery stream**（创建传输流）。 1. 从 **Amazon Data Firehose 传输流页面中，**选择您刚刚创建的传输流（）**ticketUploadStream**。在 **Details**（详细信息）选项卡上，记下流的 Amazon Resource Name (ARN) 以供日后使用。 有关创建传输流的更多信息，请参阅《Amazon Data Firehose 开发人员指南》**中的[创建 Amazon Data Firehose 传输流](https://docs.aws.amazon.com/firehose/latest/dev/basic-create.html)。有关创建 IAM 角色的更多信息，请参阅 I [A *M 用户指南*中的创建角色以向 AWS 服务委派权限](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-service.html)。 您已创建具有所需权限的 Firehose 传输流。要继续，请参阅 [为 Firehose 传输流订阅 Amazon SNS 主题](firehose-example-subscribe-delivery-stream-to-topic.md)。 # 为 Firehose 传输流订阅 Amazon SNS 主题 本主题介绍如何为[消息归档和分析示例使用案例](firehose-example-use-case.md)创建以下资源： + AWS Identity and Access Management (IAM) 角色，允许订阅 Amazon SNS 将记录放入传送流。 + 到 Amazon SNS 主题的 Firehose 传输流订阅。 **要为 Amazon SNS 订阅创建 IAM 角色** 1. 打开 IAM 控制台的[角色页面](https://console.aws.amazon.com/iam/home?#/roles)。 1. 选择**创建角色**。 1. 对于**Select type of trusted entity（选择受信任实体的类型）**，选择 **AWS service（服务）**。 1. 对于 **Choose a use case**（选择使用案例），选择 **SNS**。然后选择**下一步：权限**。 1. 选择**下一步：标签**。 1. 选择**下一步：审核**。 1. 在**审核**页面上，对于**角色名称**，输入 **ticketUploadStreamSubscriptionRole**。然后选择**创建角色**。 1. 创建角色后，选择其名称 (**ticketUploadStreamSubscriptionRole**)。 1. 在角色的 **Summary**（摘要）页面上，选择 **Add inline policy**（添加内联策略）。 1. 在 **Create policy**（创建策略）页面上，选择 **JSON** 选项卡，然后将以下策略粘贴到文本框中： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Action": [ "firehose:DescribeDeliveryStream", "firehose:ListDeliveryStreams", "firehose:ListTagsForDeliveryStream", "firehose:PutRecord", "firehose:PutRecordBatch" ], "Resource": [ "arn:aws:firehose:us-east-1:123456789012:deliverystream/ticketUploadStream" ], "Effect": "Allow" } ] } ``` ------ 在本政策中，将 AWS 账户 数字 (*123456789012*) 替换为自己的数字，然后相应地更改 AWS 区域 (*us-east-1*)。 1. 选择**查看策略**。 1. 在 **Create policy**（创建策略）页面上，对于 **Name**（名称），输入 **FirehoseSnsPolicy**。然后选择**创建策略**。 1. 在角色的 **Summary**（摘要）页面上，记下**角色 ARN**供稍后使用。 有关创建 IAM 角色的更多信息，请参阅 I [A *M 用户指南*中的创建角色以向 AWS 服务委派权限](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-service.html)。 **为 Firehose 传输流订阅 SNS 主题** 1. 打开 Amazon SNS 控制台中的[主题](https://console.aws.amazon.com/sns/home#/topics)页面。 1. 在 **Subscriptions**（订阅）选项卡上，选择 **Create subscription**（创建订阅）。 1. 在**详细信息**下，对于**方案**，选择 ****。 1. 在 En **d** point 中，输入您之前创建的**ticketUploadStream**传输流的亚马逊资源名称 (ARN)。例如，输入 **arn:aws:firehose:us-east-1:123456789012:deliverystream/ticketUploadStream**。 1. 对于**订阅角色 ARN**，请输入您之前创建的 **ticketUploadStreamSubscriptionRole**IAM 角色的 ARN。例如，输入 **arn:aws:iam::123456789012:role/ticketUploadStreamSubscriptionRole**。 1. 选择 **Enable raw message delivery**（启用原始消息传输）复选框。 1. 选择**创建订阅**。 您已创建 IAM 角色和 SNS 主题订阅。要继续，请参阅 [测试和查询 Amazon SNS 配置以实现有效的数据管理](firehose-example-test-and-query.md)。 # 测试和查询 Amazon SNS 配置以实现有效的数据管理 本主题介绍如何通过将消息发布到 Amazon SNS 主题来测试[消息归档和分析示例使用案例](firehose-example-use-case.md)。这些说明包括一个示例查询，您可以运行并根据自己的需求对其进行调整。 **测试配置** 1. 打开 Amazon SNS 控制台中的[主题](https://console.aws.amazon.com/sns/home#/topics)页面。 1. 选择 **ticketTopic** 主题。 1. 选择**发布消息**。 1. 在 **Publish message to topic**（发布消息到主题）页面上，为消息正文输入以下内容。在消息结尾处添加换行符。 ``` {"BookingDate":"2020-12-15","BookingTime":"2020-12-15 04:15:05","Destination":"Miami","FlyingFrom":"Vancouver","TicketNumber":"abcd1234"} ``` 保留所有其他选项作为默认值。 1. 选择**发布消息**。 有关发布消息的更多信息，请参阅 [发布 Amazon SNS 消息](sns-publishing.md)。 1. 在 60 秒的传输流间隔后，打开 [Amazon Simple Storage Service (Amazon S3) 控制台](https://console.aws.amazon.com/s3/home)并选择您[最初创建的](firehose-example-initial-resources.md) Amazon S3 存储桶。 存储桶中会显示已发布的消息。 **要查询数据** 1. 打开 [Amazon Athena 控制台](https://console.aws.amazon.com/athena/home)。 1. 运行查询。 例如，假设 `default` schema 中的 `notifications` 表包含下列数据： ``` {"BookingDate":"2020-12-15","BookingTime":"2020-12-15 04:15:05","Destination":"Miami","FlyingFrom":"Vancouver","TicketNumber":"abcd1234"} {"BookingDate":"2020-12-15","BookingTime":"2020-12-15 11:30:15","Destination":"Miami","FlyingFrom":"Omaha","TicketNumber":"efgh5678"} {"BookingDate":"2020-12-15","BookingTime":"2020-12-15 3:30:10","Destination":"Miami","FlyingFrom":"NewYork","TicketNumber":"ijkl9012"} {"BookingDate":"2020-12-15","BookingTime":"2020-12-15 12:30:05","Destination":"Delhi","FlyingFrom":"Omaha","TicketNumber":"mnop3456"} ``` 要查找最佳目标，请运行以下查询： ``` SELECT destination FROM default.notifications GROUP BY destination ORDER BY count(*) desc LIMIT 1; ``` 要查询在特定日期和时间范围内售出的票证，请运行如下所示的查询： ``` SELECT * FROM default.notifications WHERE bookingtime BETWEEN TIMESTAMP '2020-12-15 10:00:00' AND TIMESTAMP '2020-12-15 12:00:00'; ``` 您可以根据自己的需求调整两个示例查询。有关使用 Athena 运行查询的更多信息，请参阅 *Amazon Athena 用户指南*中的[入门](https://docs.aws.amazon.com/athena/latest/ug/getting-started.html)。 ## 清理 为避免在完成测试后产生使用费用，请删除您在本教程中创建的以下资源： + Amazon SNS 订阅 + 亚马逊 SNS 主题 + Amazon Simple Queue Service（Amazon SQS）队列 + 亚马逊 S3 存储桶 + 传输流 + AWS Identity and Access Management (IAM) 角色和策略 # 使用模板自动存档 Amazon SNS 消息 CloudFormation 要自动部署 Amazon SNS [消息存档和分析示例使用案例](firehose-example-use-case.md)，您可以使用以下 YAML 模板： ``` --- AWSTemplateFormatVersion: '2010-09-09' Description: Template for creating an SNS archiving use case Resources: ticketUploadStream: DependsOn: - ticketUploadStreamRolePolicy Type: AWS::KinesisFirehose::DeliveryStream Properties: S3DestinationConfiguration: BucketARN: !Sub 'arn:${AWS::Partition}:s3:::${ticketArchiveBucket}' BufferingHints: IntervalInSeconds: 60 SizeInMBs: 1 CompressionFormat: UNCOMPRESSED RoleARN: !GetAtt ticketUploadStreamRole.Arn ticketArchiveBucket: Type: AWS::S3::Bucket ticketTopic: Type: AWS::SNS::Topic ticketPaymentQueue: Type: AWS::SQS::Queue ticketFraudQueue: Type: AWS::SQS::Queue ticketQueuePolicy: Type: AWS::SQS::QueuePolicy Properties: PolicyDocument: Statement: Effect: Allow Principal: Service: sns.amazonaws.com Action: - sqs:SendMessage Resource: '*' Condition: ArnEquals: aws:SourceArn: !Ref ticketTopic Queues: - !Ref ticketPaymentQueue - !Ref ticketFraudQueue ticketUploadStreamSubscription: Type: AWS::SNS::Subscription Properties: TopicArn: !Ref ticketTopic Endpoint: !GetAtt ticketUploadStream.Arn Protocol: firehose SubscriptionRoleArn: !GetAtt ticketUploadStreamSubscriptionRole.Arn ticketPaymentQueueSubscription: Type: AWS::SNS::Subscription Properties: TopicArn: !Ref ticketTopic Endpoint: !GetAtt ticketPaymentQueue.Arn Protocol: sqs ticketFraudQueueSubscription: Type: AWS::SNS::Subscription Properties: TopicArn: !Ref ticketTopic Endpoint: !GetAtt ticketFraudQueue.Arn Protocol: sqs ticketUploadStreamRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Version: '2012-10-17 ' Statement: - Sid: '' Effect: Allow Principal: Service: firehose.amazonaws.com Action: sts:AssumeRole ticketUploadStreamRolePolicy: Type: AWS::IAM::Policy Properties: PolicyName: FirehoseticketUploadStreamRolePolicy PolicyDocument: Version: '2012-10-17 ' Statement: - Effect: Allow Action: - s3:AbortMultipartUpload - s3:GetBucketLocation - s3:GetObject - s3:ListBucket - s3:ListBucketMultipartUploads - s3:PutObject Resource: - !Sub 'arn:aws:s3:::${ticketArchiveBucket}' - !Sub 'arn:aws:s3:::${ticketArchiveBucket}/*' Roles: - !Ref ticketUploadStreamRole ticketUploadStreamSubscriptionRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Version: '2012-10-17 ' Statement: - Effect: Allow Principal: Service: - sns.amazonaws.com Action: - sts:AssumeRole Policies: - PolicyName: SNSKinesisFirehoseAccessPolicy PolicyDocument: Version: '2012-10-17 ' Statement: - Action: - firehose:DescribeDeliveryStream - firehose:ListDeliveryStreams - firehose:ListTagsForDeliveryStream - firehose:PutRecord - firehose:PutRecordBatch Effect: Allow Resource: - !GetAtt ticketUploadStream.Arn ```