什麼是 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 度職業支持、個性化指導、同行交流機會以及更多幫助您掌握機器學習和人工智能的內容。