で Kinesis Data Streams と Firehose を使用して DynamoDB レコードを Amazon S3 に配信する Amazon S3 AWS CDK

作成者: Shashank Shrivastava (AWS) と Daniel Matuki da Cunha (AWS）

コードリポジトリ: Amazon DynamoDB の Amazon S3 への取り込み	環境：PoC またはパイロット	テクノロジー：サーバーレス、データレイク、データベース、ストレージとバックアップ
AWS サービス: AWS CDK、Amazon DynamoDBAmazon Data Firehose、Amazon Kinesis Data Streams、AWSLambda、Amazon S3

[概要]

このパターンは、Amazon Kinesis Data Streams と Amazon Data Firehose を使用して Amazon DynamoDB から Amazon Simple Storage Service (Amazon S3) にレコードを配信するためのサンプルコードとアプリケーションを提供します。このパターンのアプローチでは AWS Cloud Development Kit (AWS CDK) L3 コンストラクトを使用し、Amazon Web Services (AWS) クラウド上のターゲット S3 バケットにデータが配信される AWS Lambda 前にでデータ変換を実行する方法の例を示します。

Kinesis Data Streams は、DynamoDB テーブルの項目レベルの変更を記録し、それらを必須の Kinesis Data Streams にレプリケートします。アプリケーションは Kinesis Data Streams にアクセスし、項目レベルの変更をほぼリアルタイムで表示できます。Kinesis Data Streams は、Firehose や Amazon Managed Service for Apache Flink などの他の Amazon Kinesis サービスへのアクセスも提供します。つまり、リアルタイムのダッシュボードの提供、アラートの生成、動的な料金設定、広告の実装、高度なデータ分析の実行を行うアプリケーションを構築できます。

このパターンは、データ統合のユースケースに使用できます。たとえば、輸送車両や産業機器は、DynamoDB テーブルに大量のデータを送信できます。その後、このデータを変換して Amazon S3 でホストされているデータレイクに保存できます。その後、Amazon Athena 、Amazon Redshift Spectrum、Amazon Rekognition 、などのサーバーレスサービスを使用して、データをクエリして処理し、潜在的な欠陥を予測できます AWS Glue。

前提条件と制限

前提条件

アクティブな AWS アカウント。
AWS Command Line Interface （AWS CLI）、インストールおよび設定済み。詳細については、 AWS CLI ドキュメントの「の開始 AWS CLI方法」を参照してください。
Node.js (18.x+) と npm、インストールおよび設定済み。詳細については、npmドキュメントの Node.js と npm のダウンロードとインストールを参照してください。
aws-cdk (2.x+)、インストールおよび設定済み。詳細については、 AWS CDK ドキュメントの「の開始 AWS CDK方法」を参照してください。
ローカルマシンでクローンおよび設定されている GitHub aws-dynamodb-kinesisfirehose-s3 取り込みリポジトリ。
DynamoDB テーブルの既存のサンプルデータ。データは以下のフォーマットを使用する必要があります：{"SourceDataId": {"S": "123"},"MessageData":{"S": "Hello World"}}

アーキテクチャ

次の図は、Kinesis Data Streams と Firehose を使用して DynamoDB から Amazon S3 にレコードを配信するワークフローの例を示しています。

この図表は、次のワークフローを示しています：

データは、DynamoDB のプロキシとして Amazon API Gateway を使用して取り込まれます。他のソースを使用して DynamoDB にデータを取り込むこともできます。
アイテムレベルの変更は、Kinesis Data Streams でほぼリアルタイムで生成され、Amazon S3 に配信されます。
Kinesis Data Streams は、変換と配信のためにレコードを Firehose に送信します。
Lambda 関数は、レコードを DynamoDB レコード形式から JSON形式に変換します。この形式には、レコード項目の属性名と値のみが含まれます。

ツール

AWS のサービス

AWS Cloud Development Kit (AWS CDK) は、コードで AWS クラウドインフラストラクチャを定義およびプロビジョニングするのに役立つソフトウェア開発フレームワークです。
AWS CDK Toolkit は、 AWS CDK アプリケーションの操作に役立つコマンドラインクラウド開発キットです。
AWS Command Line Interface （AWS CLI）は、コマンドラインシェルのコマンド AWS のサービスを通じてとやり取りするのに役立つオープンソースツールです。
AWS CloudFormation は、 AWS リソースをセットアップし、迅速かつ一貫してプロビジョニングし、 AWS アカウントと全体のライフサイクルを通じてリソースを管理するのに役立ちます AWS リージョン。

コードリポジトリ

このパターンのコードは GitHub aws-dynamodb-kinesisfirehose-s3 つの取り込みリポジトリにあります。

エピック

タスク説明必要なスキル

タスク	説明	必要なスキル
SDK の依存関係をインストールします。	ローカルマシンで、次のコマンドを実行して、`package.json`、`pattern/aws-dynamodb-kinesisstreams-s3`、および `sample-application` ディレクトリのファイルから依存関係をインストールします。 `cd <project_root>/pattern/aws-dynamodb-kinesisstreams-s3` `npm install && npm run build` `cd <project_root>/sample-application/` `npm install && npm run build`	アプリ開発者、一般 AWS
CloudFormation テンプレートを生成します。	`cd <project_root>/sample-application/` コマンドを実行します。 `cdk synth` コマンドを実行して CloudFormation テンプレートを生成します。 `AwsDynamodbKinesisfirehoseS3IngestionStack.template.json`出力は`cdk.out` ディレクトリに保存されています。 AWS CDK または AWS Management Console を使用して、でテンプレートを処理します CloudFormation。	アプリ開発者、一般 AWS、 AWS DevOps

SDK の依存関係をインストールします。

ローカルマシンで、次のコマンドを実行して、package.json、pattern/aws-dynamodb-kinesisstreams-s3、および sample-application ディレクトリのファイルから依存関係をインストールします。


cd <project_root>/pattern/aws-dynamodb-kinesisstreams-s3


npm install && npm run build


cd <project_root>/sample-application/


npm install && npm run build

アプリ開発者、一般 AWS

CloudFormation テンプレートを生成します。

cd <project_root>/sample-application/ コマンドを実行します。
cdk synth コマンドを実行して CloudFormation テンプレートを生成します。
AwsDynamodbKinesisfirehoseS3IngestionStack.template.json出力はcdk.out ディレクトリに保存されています。
AWS CDK または AWS Management Console を使用して、でテンプレートを処理します CloudFormation。

アプリ開発者、一般 AWS、 AWS DevOps

タスク	説明	必要なスキル
リソースを確認してデプロイしてください。	`cdk diff` コマンドを実行して、 AWS CDK コンストラクトによって作成されるリソースタイプを識別します。リソースをデプロイするには、`cdk deploy` コマンドを実行します。	アプリ開発者、一般 AWS、 AWS DevOps

タスク説明必要なスキル

タスク	説明	必要なスキル
DynamoDB テーブルにサンプルデータを取り込みます。	で次のコマンドを実行して、DynamoDB テーブルにリクエストを送信します AWS CLI。 `aws dynamodb put-item --table-name <your_table_name> --item '{"<table_partition_key>": {"S": "<partition_key_ID>"},"MessageData":{"S": "<data>"}}'` 例： `aws dynamodb put-item --table-name SourceData_table --item '{"SourceDataId": {"S": "123"},"MessageData":{"S": "Hello World"}}'` デフォルトでは、`put-item` オペレーションが成功しても、は出力として値を返しません。操作が失敗した場合はエラーを返します。データは DynamoDB に保存され、Kinesis Data Streams と Firehose に送信されます。注：DynamoDB テーブルにデータを追加するには、さまざまな方法を使用します。詳細については、DynamoDB ドキュメントの「テーブルへのデータのロード」を参照してください。	アプリ開発者
S3 バケットに新しいオブジェクトが作成されたことを確認します。	にサインイン AWS Management Console し、S3 バケットをモニタリングして、送信したデータで新しいオブジェクトが作成されたことを確認します。詳細については、Amazon S3 ドキュメントのGetObject「」を参照してください。	アプリ開発者、一般 AWS

DynamoDB テーブルにサンプルデータを取り込みます。

で次のコマンドを実行して、DynamoDB テーブルにリクエストを送信します AWS CLI。

aws dynamodb put-item --table-name <your_table_name> --item '{"<table_partition_key>": {"S": "<partition_key_ID>"},"MessageData":{"S": "<data>"}}'

例：

aws dynamodb put-item --table-name SourceData_table --item '{"SourceDataId": {"S": "123"},"MessageData":{"S": "Hello World"}}'

デフォルトでは、put-item オペレーションが成功しても、は出力として値を返しません。操作が失敗した場合はエラーを返します。データは DynamoDB に保存され、Kinesis Data Streams と Firehose に送信されます。

注：DynamoDB テーブルにデータを追加するには、さまざまな方法を使用します。詳細については、DynamoDB ドキュメントの「テーブルへのデータのロード」を参照してください。

アプリ開発者

S3 バケットに新しいオブジェクトが作成されたことを確認します。

にサインイン AWS Management Console し、S3 バケットをモニタリングして、送信したデータで新しいオブジェクトが作成されたことを確認します。

詳細については、Amazon S3 ドキュメントのGetObject「」を参照してください。