作者：伊藤雅博，日立製作所股份有限公司。

首先

在AWS上实现流处理时，一般的做法是将数据排队的消息服务与处理排队数据的流处理服务结合使用。本篇文章将介绍AWS提供的各种消息服务中，以下服务在处理流数据时经常被充分利用的特点。

• Amazon Kinesis Data Streams (KDS)

 

• Amazon Managed Streaming for Apache Kafka (MSK)

 

Amazon Simple Queue Service (SQS)

顺便提一下，此篇文章的内容基于2020年中期的调查结果，并进行了一些更新。由于AWS的服务规范会随时更新，因此可能与最新规范有所不同。请参阅AWS的官方文件以获取最新信息。

投稿一覧：
1. 处理流数据的AWS消息服务
2. AWS流处理的特点，包括KDS（Kinesis）、MSK（Kafka）、SQS（消息队列服务）
3. AWS的Kinesis Data Analytics、EMR Spark Streaming、Lambda进行流处理的特点

在AWS中的流处理的基本结构。

以下的图表显示了构成流处理系统的AWS服务组合。

在流式处理系统中，通常会使用消息服务将流数据排队，然后进行流式处理。更多详情请参考上一篇帖子《在AWS上处理流式数据的消息服务》。

KDS、MSK、SQS的特点和架构。

以下是关于AWS上处理流式数据时常用的KDS、MSK、SQS的特点和架构的介绍。

亚马逊 Kinesis 数据流 (KDS)

KDS的典型相关组件。

以下的图示了KDS的典型相关组件。KDS使用客户端库，可以从各种语言的应用程序中使用。此外，还可以与AWS的各种服务进行协作。请注意，由于随时会添加与KDS对应的服务，因此请查阅AWS的官方文件以获取最新的支持情况。

KDS的架构

KDS的架构如下图所示。KDS是一种分布式消息队列，采用发布/订阅模式。KDS将一个虚拟队列称为Stream，每个Stream由多个Shard组成。通过将写入/读取操作分散到多个Shard上，可以实现性能的扩展。

在KDS中写入的数据是以Key-Value形式记录的，而且Value部分可以存储任意的二进制数据（Data BLOB）。

当从KDS中读取记录时，读取方的应用程序会轮询Shard以获取记录。获取采用的是Pull方式，KDS不能进行Push传输。但是，如果使用AWS Lambda，可以使用写入Stream作为触发器来启动Lambda函数，从而构建基于事件驱动的应用程序来获取和处理记录。

而且，由於Record在讀取時不會被刪除，因此可以從多個應用程序中並行讀取同一個Stream的數據。

KDS处理的流程

以下是展示KDS处理流程的图示。这里以使用Kinesis Producer Library (KPL)进行写入，使用Kinesis Client Library (KCL)进行读取的例子进行说明。KPL/KCL可以嵌入到任何用户应用程序中并进行使用。

当用户应用程序想要将要发送的记录注册到 KPL 中时，KPL 会将记录发送到目标数据流的分片中。记录的写入目标分片是根据记录的分区键的散列值来确定的。

而且，KPL还提供以下所示的高级写入功能。

• 複数Recordを1つのRecordに集約して送信

 

• 書き込み失敗時の自動再試行（リトライ）

 

CloudWatchへの各種メトリクス送信（送信クライアント監視）

在使用KCL的读取端应用程序中，可以并行读取每个Shard的Record。每个应用程序会在读取端应用程序自身而不是KDS中管理自己读取了多少Record。具体来说，读取端应用程序需要在记忆中保存指示从Shard读取了多少Record的序列号。

KCL提供了高级的读取功能，如下所示。

• KPLが集約したRecordの解凍

 

• 読み出し失敗時の自動再試行（リトライ）

 

• CloudWatchへの各種メトリクス送信（受信クライアント監視）

 

読み出し位置(シーケンス番号)のDynamoDBへの自動記録

KDS的规格和限制

以下的图表展示了KDS的主要规格和限制。

可以将1条记录（BLOB）大小最大为1MB的数据写入Stream。写入操作的最大吞吐量受Stream的Shard数量限制。每个Shard的写入吞吐量最大为1MB/秒或1,000条记录/秒，而Stream内的Shard数量最多可以增加到10,000个。换句话说，写入吞吐量最大可以扩展到10,000MB/秒。需要注意的是，超出吞吐量限制的写入请求将被拒绝，因此需要进行重试。当使用KPL时，会自动进行重试。

为了发挥KDS的写入性能，需要调整Shard数量和设计分区键等进行优化。由于最大吞吐量取决于Shard数量，如果Shard数量不足或由于分区键偏差导致写入Shard偏差，将导致频繁重试请求和吞吐量下降。

在 Stream 中写入的 Record 会被保存一定时间，并且即使被读取也不会被删除。Record 的保存时间最长为 365 天。保存的 Record 通过同步复制到三个可用性区域（AZ）来确保可用性。

通过Stream读取的吞吐量受Stream的分片数量限制。每个分片的读取吞吐量最大为2MB/秒或2,000条记录/秒。由于一个Stream内最多有10,000个分片，所以读取吞吐量最大可以扩展到20,000MB/秒。

请注意，如果超过限制的读取请求将被拒绝，因此需要进行重试。在使用KCL时，会自动进行重试。每个读取请求最多可以获取10,000条记录或10MB的数据。例如，如果在一次读取请求中获取了10MB的数据，那么在接下来的5秒内将无法发送其他请求。

亚马逊托管的Apache Kafka流 (MSK)

MSK的架构 (MSK de jià gé)

以下图示了MSK的架构。MSK在集群中构建了一个分布式发布/订阅型消息队列。

请用多台Kafka Broker实例组成MSK集群，集群上配置一个名为Topic的分布式消息队列。MSK的Broker会自动建立，并且可以扩展到最大100台。

使用Kafka Producer库进行向主题写入数据，使用Kafka Consumer库进行读取数据。Producer/Consumer库支持Java、C/C++、Python等多种语言，可以嵌入到任意的用户应用程序或物联网设备中。

在MSK中，ZooKeeper用于Broker之间的协作，因其会自动构建，所以无需管理。

MSK处理的流程

我們將以下圖表顯示MSK處理的流程。

主题由一个或多个分区组成，每个分区由一个Leader副本和零个或多个Follower副本组成。Leader和Follower分别部署在不同的Broker实例上。数据的写入/读取只能对Partition的Leader副本进行。

在Topic中添加的数据是以Key-Value形式的Record表示的，可以指定任意的文本或者二进制数据。默认设置下，每个Record的最大大小是1MB。

写入方面的应用程序使用Producer库将数据分布式地写入到每个分区中。通过在Broker之间分布处理写入操作，实现横向扩展。

将写入Partition Leader副本的记录复制到另一个Broker的Follower副本中。当Broker发生故障且Leader副本无法使用时，其中一个Follower副本将被提升为Leader副本。这样可以避免部分Broker实例的故障导致停机和数据丢失。

读取端的应用程序使用Consumer库并行地从每个分区中读取记录。读取操作与KDS类似，通过应用程序端进行轮询。因此，MSK无法从自身进行推送交付。

读取方可以通过应用程序单元构建消费者组，组内的消费者可以共享一个主题的分区来进行读取。由于具有并行读取功能，因此与Spark等并行分布式处理框架兼容性良好。此外，记录从主题中读取后不会消失，因此多个应用程序（消费者组）可以同时读取同一主题的同一记录。

请参考《Apache Kafka的概述和架构》以了解有关 Kafka 的详细机制。

亚马逊简单队列服务 (SQS)

SQS的标准队列和FIFO队列的特点。

SQS有两种类型的队列：高性能标准队列和保证消息顺序的FIFO队列。下表列出了它们各自的特点。

如果您希望严格保持消息的顺序，或者需要避免重复发送的情况，我们建议使用FIFO队列。否则，我们建议使用标准队列。

SQS的架构和处理流程

下图展示了SQS的规范和处理流程。这是使用SQS客户端库Producer/Consumer的一个示例。

在SQS中写入的消息由唯一的ID、最多10个元数据属性和最大为256KB的文本数据有效负载组成。

在书写应用程序使用SQS的Producer库发送消息到队列上（发送请求）。在此过程中，每次最多可以发送10个消息或256KB的消息。消息将保存在队列中并在单一区域内的多个可用区进行复制。可以设置保存时间为1分钟至14天（默认为4天），并且可以保存无限数量的消息。

阅读端应用程序使用SQS的消费者库来轮询队列获取消息（接收请求）。当队列中没有消息时，可以通过长轮询最多20秒来立即接收新到的消息。可以一次接收最多10条或256 KB的消息。

消费者从队列中接收的消息会被锁定30秒钟，其他消费者无法接收。读取端应用程序需要在30秒内处理消息。处理消息成功后，应通过指定已接收消息的ID向SQS发出删除指令（删除请求）。每次最多可以删除10条或256 KB的消息。转发中（读取端已接收但未删除）的消息数量有限制（标准队列：120,000条，FIFO队列：20,000条），注意必须删除已接收消息，否则队列会溢出。

在SQS中，当一个客户端读取一条消息时，需要对其进行锁定并删除。因此，无法像KDS或MSK那样从多个应用程序中读取相同的消息。

处理流数据时，KDS、MSK、SQS具有不同的特点。

以下表格展示了KDS、MSK和SQS的特点。

以下是使用这些服务处理流数据时的优点和注意事项。

亚马逊Kinesis数据流 (KDS)

• メリット：

AWSサービスとの親和性が高い
Shardという並列数の概念があるため、分散ストリーム処理サービス（KDA、EMR、Lambda）と相性がよい
並列数をユーザが制御可能なので、最初から最大性能を出しやすい

注意点：

性能を引き出すにはチューニングが必要（特にShard数の調整とパーティションキーの設計）

読み書きスループットがShard数に依存するため、Shard数が不足している場合や、パーティションキーの偏りによる書き込み先Shardの偏りが発生すると、リクエストの再試行が頻発してスループットが低下するので注意

亚马逊托管的Apache Kafka流媒体服务（MSK）

虽然具有与KDS几乎相同的特点，但以下有一些区别。

• メリット：

OSSのKafkaベースなのでベンダーロックインを回避しやすい
Partition単位のスループット制限がないので、KDSよりは若干チューニングしやすい

注意点：

マネージドサービスであるがインスタンス台数などを意識する必要がある

亚马逊简单队列服务 (SQS)

• メリット：

チューニング不要で簡単に使える

注意点：

テキストデータ、かつ256KB以下のデータしか扱えない
同じデータを複数の宛先に配信できない（読み出したデータは即時削除する必要があるため）
対応するストリーム処理サービスはLambdaのみ。性能（Lambda同時実行数）は自動スケールで制御不可

最后

我們介紹了AWS提供的幾種消息服務中，經常被用於處理流數據的特點和架構，包括KDS、MSK和SQS。MSK和KDS適合結合流處理使用，但需要進行一些調優才能發揮性能。而SQS的特點是無需進行調優，可以輕鬆使用。

在下一篇帖子中，我們將介紹與本次提到的訊息服務、Kinesis Data Analytics、EMR的Spark Streaming和Lambda相結合的流式處理特點。

第3回: AWSのKinesis Data Analytics、EMR Spark Streaming、Lambdaによるストリーム処理の特徴