Amazon Kinesis Data Streams
开发人员指南
AWS 文档中描述的 AWS 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅 Amazon AWS 入门

教程:使用 Amazon Kinesis Data Streams 可视化 Web 流量

本教程通过介绍关键 Kinesis Data Streams 结构(具体而言就是数据创建器数据使用者)来帮助您开始使用 Amazon Kinesis Data Streams。本教程使用了基于实时数据分析的常用案例的示例应用程序,如什么是 Amazon Kinesis Data Streams? 中所述。

此示例的 Web 应用程序使用一个简单的 JavaScript 应用程序来轮询 DynamoDB 表,该表用于存储滑动窗口内前 N 个分析的结果。该应用程序将采用此数据并创建结果的可视化效果。

Kinesis Data Streams 数据可视化示例应用程序

本教程中的数据可视化示例应用程序演示如何使用 Kinesis Data Streams 进行实时数据注入和分析。该示例应用程序将创建一个数据创建器,该数据创建器会将来自各个 URL 的模拟访问者计数放入一个 Kinesis data stream。该流将按照接收这些数据记录的顺序长期存储它们。数据使用者从该流中获取这些记录,然后计算有多少访问者来自某个特定 URL。最后,会有一个简单的 Web 应用程序实时轮询结果以提供可视化计算结果。

此示例应用程序将演示在 10 秒期间内执行滑动窗口分析的常用流处理使用案例。上述可视化中显示的数据以持续更新的图表的形式反映流的滑动窗口分析结果。此外,数据使用者将对数据流执行 Top-K 分析,以计算在计数上排名前三的引用站点,它们将立即显示在图表下方的表格中,并且每两秒更新一次。

为了让您快速入门,该示例应用程序使用了 AWS CloudFormation。利用 AWS CloudFormation,您可以创建模板来描述 AWS 资源以及运行应用程序所需的任何关联依赖项或运行时参数。该示例应用程序使用模板快速创建所有必要的资源,包括在 Amazon EC2 实例上运行的创建器和使用者应用程序以及 Amazon DynamoDB 中用来存储聚合记录计数的表。

注意

示例应用程序在启动后将因使用 Kinesis Data Streams 而产生象征性的费用。在可能的情况下,示例应用程序将使用有资格享受 AWS 免费套餐的资源。当您完成本教程时,请删除 AWS 资源以停止产生费用。有关更多信息,请参阅 步骤 3:删除示例应用程序

先决条件

本指南将帮助您设置、运行和查看 Kinesis Data Streams 数据可视化示例应用程序的结果。要开始使用示例应用程序,您首先需要执行以下操作:

步骤 1:启动示例应用程序

使用 AWS 提供的 AWS CloudFormation 模板启动示例应用程序。该示例应用程序有一个流写入器(随机生成记录并将其发送到 Kinesis data stream)、一个数据使用者(将针对某个资源的 HTTPS 请求进行计数)和一个 Web 应用程序(以持续更新的图表的形式显示流处理数据的输出)。

启动应用程序

  1. 打开此教程的 AWS CloudFormation 模板

  2. Select Template 页面上,提供了此模板的 URL。选择 Next

  3. 请注意,Specify Details 页面上的默认实例类型是 t2.micro。但是,T2 实例要求有一个 VPC。如果您的 AWS 账户在您的区域没有默认 VPC,则必须将 InstanceType 更改为其他实例类型,如 m3.medium。选择 Next

  4. Options 页面上,您可以选择键入标签键和标签值。此标签将添加到基于模板创建的资源,如 EC2 实例。选择 Next

  5. Review page 上,选择 I acknowledge that this template might cause AWS CloudFormation to create IAM resources,然后选择 Create

最初,您应该会看到名为 KinesisDataVisSample 且状态为 CREATE_IN_PROGRESS 的堆栈。该堆栈可能需要几分钟时间来创建。当该状态为 CREATE_COMPLETE 时,继续执行下一步。如果状态未更新,请刷新页面。

步骤 2:查看示例应用程序的组件

Kinesis Data Stream

能够实时注入来自大量创建器的数据、长期存储这些数据,并将这些数据提供给多个使用者。流表示数据记录的有序序列。创建流时,必须指定流名称和分片计数。一个流包含一个或多个分片;每个分片是一组数据记录。

AWS CloudFormation 会为示例应用程序自动创建流。AWS CloudFormation 模板的此部分显示了在 CreateStream 操作中使用的参数。

查看堆栈详细信息

  1. 选择 KinesisDataVisSample 堆栈。

  2. Outputs 选项卡中,选择 URL 中的链接。URL 的格式应类似于 http://ec2-xx-xx-xx-xx.compute-1.amazonaws.com。

  3. 在大约 10 分钟后,应用程序堆栈才能创建完成,数据分析图表中才会显示有意义的数据。实时数据分析图将显示在名为 Kinesis Data Streams Data Visualization Sample 的单独页面上。它显示了引用 URL 在 10 秒的时间范围内发送的请求数,该图表每 1 秒更新一次。图表的时间范围是最近 2 分钟。

    示例应用程序的数据可视化

查看流详细信息

  1. 通过以下网址打开 Kinesis 控制台:https://console.amazonaws.cn/kinesis

  2. 选择名称为以下格式的流:KinesisDataVisSampleApp-KinesisStream-[randomString]

  3. 选择流名称以查看流的详细信息。

  4. 这些图表显示了将记录放入流中的数据创建器以及从流中获取数据的数据使用者的活动。

     示例应用程序的流监控图表

数据创建器

数据创建器向 Kinesis data stream 提交数据记录。为将数据放入流中,创建器会对流调用 PutRecord 操作。

每个 PutRecord 调用都需要流名称、分区键和创建器正在添加到流的数据记录。流名称用于确定记录所在的流。分区键用于确定数据记录要添加到的流中的分片。

使用哪个分区键取决于您的应用程序逻辑。在大多数情况下,分区键的数量应比分片的数量多得多。分区键的数量多于分片的数量可让流将数据记录均匀地分布在流中的各个分片中。

数据创建器使用六个常用 URL 作为放入到包含两个分片的流中的每个记录的分区键。这些 URL 代表模拟的页面访问。HttpReferrerKinesisPutter 代码的 99-132 行会将数据发送到 Kinesis Data Streams。三个必需参数是在调用 PutRecord 之前设置的。分区键是使用 pair.getResource 设置的,该命令将随机选择在 HttpReferrerStreamWriter 代码的 85-92 行中创建的六个 URL 之一。

数据创建器可以是将数据放入 Kinesis Data Streams 的任何东西,如 EC2 实例、客户端浏览器或移动设备。示例应用程序对其数据创建器和数据使用者使用一个 EC2 实例;然而,在大多数现实世界情景中,示例应用程序的每个组件都有单独的 EC2 实例。您可以按照以下指示从示例应用程序查看 EC2 实例数据。

在控制台中查看实例数据

  1. 打开 Amazon EC2 控制台 https://console.amazonaws.cn/ec2/

  2. 在导航窗格中,选择 Instances

  3. 选择为示例应用程序创建的实例。如果您不能确定是哪个实例,请注意它的安全组名称以 KinesisDataVisSample 开头,您可以据此特征判断。

  4. Monitoring 选项卡上,您会看到示例应用程序的数据创建器和数据使用者对资源的使用情况。

数据使用者

数据使用者从 Kinesis data stream 的分片中检索和处理数据记录。每个使用者都从一个特定分片中读取数据。使用者使用 GetShardIteratorGetRecords 操作从分片中获取数据。

分片迭代器代表了使用者将从中读取数据的流和分片的位置。当使用者开始从流中读取记录时或更改从流中读取记录的位置时,它将获得一个分片迭代器。要获取分片迭代器,您必须提供流名称、分片 ID 和分片迭代器类型。利用分片迭代器类型,使用者可以指定要从流中的哪个位置开始读取记录(例如从数据实时到达的流的开头)。流批量返回记录,您可以使用可选的限制参数控制其大小。

数据使用者在 DynamoDB 中创建表来维护应用程序的状态信息(如检查点和工作程序分片映射)。每个应用程序都有自己的 DynamoDB 表。

数据使用者对最近两秒来自每个特定 URL 的访问者请求进行计数。此类型的实时应用程序对滑动窗口采用 Top-N 分析。在这种情况下,Top-N 是访问者请求的前三页,而滑动窗口为两秒。这是常见的处理模式,体现了现实情况下使用 Kinesis Data Streams 进行的数据分析。此计算的结果保存在 DynamoDB 表中。

查看 Amazon DynamoDB 表

  1. 通过以下网址打开 DynamoDB 控制台:https://console.amazonaws.cn/dynamodb/

  2. 在导航窗格中,选择 Tables

  3. 示例应用程序创建了两个表:

    • KinesisDataVisSampleApp-KCLDynamoDBTable-[randomString] - 管理状态信息。

    • KinesisDataVisSampleApp-CountsDynamoDBTable-[randomString] - 保存滑动窗口内前 N 次分析的结果。

  4. 选择 KinesisDataVisSampleApp-KCLDynamoDBTable-[randomString] 表。在表中有两个条目,指示特定分片 (leaseKey)、流中的位置 (checkpoint) 和读取数据的应用程序 (leaseOwner)。

  5. 选择 KinesisDataVisSampleApp-CountsDynamoDBTable-[randomString] 表。您可以查看聚合的访问者计数 (referrerCounts),该计数是数据使用者作为滑动窗口分析的一部分计算的。

Kinesis 客户端库 (KCL)

使用者应用程序可使用 Kinesis 客户端库 (KCL) 来简化流的并行处理。KCL 负责许多与分布式计算相关的复杂任务,例如对多个实例进行负载均衡、对实例故障做出响应、对已处理的数据执行检查点操作和对重新分片做出反应。KCL 可让您将精力放在编写记录处理逻辑上。

数据使用者向 KCL 提供它要在流中开始读取数据的位置,在本例中是指定流开头尽可能最新的数据。库使用此数据来代表使用者调用 GetShardIterator。使用者组件还向客户端库提供了要对使用名为 IRecordProcessor 的重要 KCL 接口处理的记录执行的操作。KCL 代表使用者调用 GetRecords,然后按照 IRecordProcessor 的指示处理这些数据。

  • HttpReferrerCounterApplication 示例代码的 92-98 行配置 KCL。这将使用库的初始配置来设置库,例如设置要在其中读取数据的流的位置。

  • HttpReferrerCounterApplication 示例代码的 104-108 行告知 KCL 在使用重要的客户端库组件 IRecordProcessor 处理记录时要使用的逻辑。

  • CountingRecordProcessor 示例代码的 186-203 行包含使用 IRecordProcessor 的 Top-N 分析的计数逻辑。

步骤 3:删除示例应用程序

示例应用程序创建两个分片,从而在应用程序运行时产生分片使用费用。为确保您的 AWS 账户不继续产生费用,请在使用完示例应用程序后删除您的 AWS CloudFormation 堆栈。

删除应用程序资源

  1. 通过以下网址打开 AWS CloudFormation 控制台:https://console.amazonaws.cn/cloudformation

  2. 选择堆栈。

  3. 依次选择 ActionsDelete Stack

  4. 当系统提示进行确认时,选择 Yes, Delete

在 AWS CloudFormation 清除与示例应用程序关联的资源时,状态会更改为 DELETE_IN_PROGRESS。当 AWS CloudFormation 清除完资源时,它会从列表中移除该堆栈。

步骤 4:后续步骤