什么是 AWS Kinesis？ 设计模式、用例和比较

我们生活在跨应用程序集成、即时通知和即时数据更新的时代。 在这种情况下，创建、维护和修改实时系统变得更加重要。

多年来，已经开发了各种有用的工具来帮助构建和维护这样的跨平台系统。 RabbitMQ、Kafka 和 AWS Kinesis 是三个这样的工具，它们帮助开发人员和工程师无缝地处理实时数据。 这些系统都被创建和维护，牢记不同的目标。 因此，根据手头的工作，它们具有独特的优势和局限性。

本文将详细讨论 AWS Kinesis 及其工作原理。

Kinesis 是建立在 AWS 之上的流媒体服务。 它可用于处理各种数据——从日志、物联网数据、视频数据，基本上是任何数据格式。 这使您可以在数据流经您的系统时实时运行不同的机器学习模型和流程。 因此，它减少了通过传统数据库的麻烦，同时提高了整体效率。

探索我们的热门 MBA 课程

Pub/Sub 设计模式

在我们深入了解 Kinesis 的具体使用方式之前，有必要更多地了解它使用的设计模型。 在这种情况下，我们谈论的是发布者和订阅者设计，这通常被称为发布/订阅设计模式。 开发此设计模式是为了让发布者（消息的发送者）将事件推送到 Kinesis（事件总线）中。 然后，此事件总线成功地将输入数据分发给所有订阅者。

这里要记住的一个关键因素是，发布者基本上不知道存在任何订阅者。 所有消息传递和消息传输完全由 AWS Kinesis 管理。

换句话说，发布/订阅设计模式用于消息的高效通信，而无需创建高度耦合的设计。 相反，Kinesis 专注于利用独立组件并以此构建整体分布式工作流程。

从本质上讲，AWS Kinesis 是一种功能强大的流媒体工具，具有明显的优势，尤其是与其他实时流媒体工具相比。 其中一个好处是它是一种托管服务，因此开发人员不必处理系统管理。 这允许开发人员更多地关注他们的代码和系统，而不是管理职责。

现在，让我们看一下 Kinesis 的一些用例。

AWS Kinesis 使用案例 – 流数据

AWS Kinesis 对于希望在不同平台上管理和集成数据的大小公司非常有用。 对于希望跨平台管理和集成数据的组织而言，Kinesis 在大规模和小型场景中都是有益的。

让我们看一下公司使用 AWS Kinesis 无缝管理大量实时数据的两个大型用例。

网飞

Netflix 每天使用 AWS Kinesis 处理多 TB 的日志数据。 Netflix 需要一个集中式应用程序来实时记录所有数据。 通过使用 Kinesis，Netflix 开发了 Dredge，它使用元数据实时丰富内容。 这样，数据在通过 Kinesis 时会立即得到处理。 这消除了将数据加载到数据库中以供将来处理的繁琐步骤。

维里通

Veriton 提供人工智能和机器学习服务。 它使用 AWS Kinesis 视频流来处理客户数据。 Veriton 还将 ML 模型和 AI 实时应用于内容，以通过指标和元数据对其进行改进。 使用这些附加信息，Veritone 可以通过查看音频、面部识别、标记数据等来更轻松地搜索 Kinesis 视频流。

这些只是当今公司如何利用 AWS Kinesis 更有效地处理实时流数据的众多示例中的两个。

让我们继续讨论 AWS Kinesis 流的技术细节和基本组件。

学习世界顶尖大学的 AI 和 ML 课程。 获得硕士、Executive PGP 或高级证书课程以加快您的职业生涯。

流与 Firehose

AWS Kinesis 为开发人员提供了两种主要产品——Kinetic Streams 和 Kinesis Firehose。

要使用 Kinesis Stream，您需要使用 Kinesis Producer Library。 它将允许您将所有实时数据放入您的流中。 此外，您可以将此库连接到几乎任何应用程序或进程。 但是，Kinesis Streams 不是 100% 托管服务。 因此，开发团队将需要在需要时手动扩展它。 此外，输入流中的数据将在那里保留 7 天。

Kinesis Firehose 实现起来稍微简单一些。 馈送到 Kinesis Firehose 的数据被发送到 Amazon Redshift、Amazon S3 甚至 Elasticsearch——所有这些都使用 AWS Kinesis 引擎。 在此之后，您可以根据您的要求对其进行处理。 如果数据存储在 Amazon S3 或任何其他 AWS 存储系统中，您可以将其保留多久，只要您愿意。

在 Kinesis 上设置流

在开始访问 Kinesis 之前，您必须通过访问 AWS CLI 来设置流。 在命令外壳中，输入以下命令以创建一个名为 DataProcessingStream 的流

--流名称数据处理流\

–分片计数 1 \

–区域 eu-west-1

使用 Python 创建流式管道

在 Kinesis 上设置流后，您必须开始构建生产者和消费者。 Kinesis 的核心组件可帮助您创建访问层以集成其他系统、软件和应用程序。

在本教程中，我们将使用 boto3 Python 库连接到 Kinesis。

创建生产者

使用下面提到的代码使用 Python 编程语言创建生产者：

导入 boto3

导入json

导入日志

logging.basicConfig(level = logging.INFO)

session = boto3.Session(region_name='eu-west-1')

客户端 = session.client('kinesis')

test_data = {'data_tag': 'DataOne', 'score': '10', 'char': '数据库勇士'}

响应 = client.put_record(

StreamName='DataProcessingStream',

数据=json.dumps({

“data_tag”：test_data['data_tag']，

“分数”：test_data['score']，

“char”：test_data['char']

}),

分区键='a01'

)

logging.info(“输入新数据分数：%s”, test_data)

要提取数据，您需要另一个脚本来监听提供给生产者的数据。 为此，您可以使用 ShardIterator 访问所有输入 Kinesis 的数据。 这样，您可以访问 Kinesis 中的实时和未来记录。

创建消费者

使用下面提到的代码创建 Python 使用者：

导入 boto3

导入json

导入系统

导入日志

logging.basicConfig(level = logging.INFO)

session = boto3.Session(region_name='eu-west-1')

客户端 = session.client('kinesis')

aws_kinesis_stream = client.describe_stream(StreamName='DataProcessingStream)

shard_id = aws_kinesis_stream['StreamDescription']['Shards'][0]['ShardId']

stream_response = client.get_shard_iterator(

StreamName='DataProcessingStream',

ShardId=shard_id,

ShardIteratorType='TRIM_HORIZON'

)

迭代器 = stream_response['ShardIterator']

而真：

尝试：

aws_kinesis_response = client.get_records(ShardIterator=iterator, Limit=5)

迭代器 = aws_kinesis_response['NextShardIterator']

在 aws_kinesis_response['Records'] 中记录：

如果记录中的“数据”和 len(record['Data']) > 0：

logging.info(“收到的新数据分数：%s”, json.loads(record['Data']))

除了键盘中断：

sys.exit()

在上面的例子中，我们只是打印出数据。

Kinesis 管道的问题

Kinesis 确实是有益的，但它并非没有挑战和缺点。 使用 Kinesis 时您将面临的重大挑战之一可以称为“可观察性”。

当您使用多个 AWS 组件时，您创建的系统将变得越来越复杂。 例如，如果您将 Lambda 函数用作生产者和消费者，并将其连接到不同的 AWS 存储系统，则管理和跟踪依赖关系和错误将变得非常困难。

阅读我们与 MBA 相关的热门文章

综上所述

毫无疑问，流数据和处理实时数据是当下的需要，而且随着我们的世界产生越来越多的数据，这种需求只会越来越多。 因此，如果您有兴趣掌握 Kinesis 的技巧，专业课程可能会有所帮助。

upGrad 的机器学习和人工智能理学硕士课程由 IIIT-B 和 LJMU 合作提供，为期 18 个月的综合课程，旨在帮助您从数据探索的基础知识开始，并掌握 NLP、Deep 等所有关键概念学习、强化学习等。 更重要的是 - 您可以参与行业项目、360 度职业支持、个性化指导、同行交流机会以及更多帮助您掌握机器学习和人工智能的内容。