チャネルコーポレーションのアーキテクチャモダナイゼーション：Amazon DynamoDB 活用事例パート 2 ストリーム

(本記事は 2024/11/06 に投稿された How チャネルコーポレーション modernized their architecture with Amazon DynamoDB, Part 2: Streams を翻訳した記事です。)

この記事は、チャネルコーポレーションの TaeHun Yoon と Changsoon Kim と共同で執筆されました。

チャネルコーポレーションは、 B2B ソフトウェア・アズ・ア・サービス（ SaaS ）スタートアップ企業で、オールインワン AI メッセンジャー「チャネルトーク」を運営しています。このシリーズの第1部では、 NoSQL 採用の動機、ビジネス成長に伴う技術的問題、そして PostgreSQL から Amazon DynamoDB への移行に関する考慮事項について紹介しました。この投稿では、 DynamoDB だけでは対処できなかった部分を解決するために、他のサービスと統合した経験を共有します。

背景：構造化データと非構造化データの取得問題

リレーショナルデータ設計と NoSQL にはいくつかの重要な違いがあります。 DynamoDB は、定義されたアクセスパターンに対する効率的なデータ取得に優れ、特定のクエリタイプに最適化されたパフォーマンスを提供します。この特徴により、次のチャネルコーポレーションのユースケースは DynamoDB だけでは解決が困難な状況でした。

様々なフィルタリング条件による構造化データの検索（メッセージ検索） – チャネルトークはユーザーが会話を素早く検索できるようにします。ユーザーは、アサイン先、チーム、フォロワーなどの様々なフィルターを追加して検索することができる必要があります。次のスクリーンショットにその例を示します。

構造化されていないデータの検索（顧客データ検索） – 顧客データは、チャネルコーポレーションの顧客が入力する情報に応じて異なるスキーマを持ちます。このスキーマは固定されておらず、執筆時点では、複数のフィールドを使用して最大 100 万件の顧客データレコードを迅速に検索できる必要があります。

チャネルコーポレーションは、これら 2 つの問題を解決するために DynamoDB と Amazon OpenSearch Service などの目的別検索サービスを使用する方が効果的で効率的であると判断しました。これを行うには、 DynamoDB と他のサービス間のデータ同期が必要です。 DynamoDB は Amazon OpenSearch Service とのゼロ ETL 統合を持っていますが、チャネルコーポレーションは現在、以下の理由で自己管理の ETL パイプラインを使用しています。

検索可能なデータに必要な最小限の情報のみを転送する必要がありました。また、特定の属性の値の変更を無視し、既存の値と現在の値の間の変更を比較する必要がありました
検索サービスで冪等性を確保するため、直接削除するのではなく、論理削除でレコードを変更および挿入する必要な場合がありました

DynamoDB とのストリームインテグレーション

時系列に並んだ一連のイベントをストリームと呼びます。DynamoDB は、ストリームとして変更データキャプチャ (CDC) を実行する 2 つの方法を提供します。

Amazon DynamoDB Streams は、DynamoDB テーブル内のアイテムレベルの変更の時系列順のシーケンスをキャプチャし、この情報を最大 24 時間ログに保存します。
Amazon Kinesis Data Streams は、DynamoDB テーブル内のアイテムレベルの変更をキャプチャし、それらを Kinesis データストリームに複製します。

チャネルコーポレーションは、これらのサービスを使用して、DynamoDB の変更データをストリームを介して検索性能の高いサービスに渡すことで、メッセージおよび顧客データ検索の問題を解決できました。
次の図は、DynamoDB Streams を使用したワークフローを示しています。

このソリューションは次の特性を提供します。

DynamoDB Streams からの読み取りは、 AWS Lambda ベースの消費者にとって無料です。
重複が削除された、時系列順のアイテムレベルの変更シーケンスを提供します。

しかし、次の点を考慮する必要があります：

Lambda での障害やストリーム処理層での障害を処理する必要があります。
開始位置が LATEST に設定されている場合、イベントを見逃す可能性があります。

次の図は、 Kinesis Data Streams を使用したワークフローを示しています。

このソリューションは次の特性を提供します：

Kinesis Data Streams と Lambda の両方のコストを含みます
最大 1 年間のデータ保持期間を設定できます
5 つの開始位置オプションを提供します

しかし、次の点を考慮する必要があります：

Lambda やストリーム処理層での障害を処理する必要があります
開始位置が LATEST に設定されている場合、イベントを見逃す可能性があります
逆のイベントや重複したイベントが発生する可能性があります

これをより詳細に理解するために、データが DynamoDB から各ストリームに渡される方法と、Lambda がストリーム上でどのように操作を実行するかを見てみましょう。

DynamoDB Streams

DynamoDB は、各パーティション内の各アイテムに対して行われた変更を対応するシャードに送信し、変更が行われた順序を維持します。このプロセスは、特定のキーが最大 1 つのシャードに存在し、その順序が維持されることを保証にします。

Kinesis Data Streams Kinesis Data Streams のデータレコードは、アイテムの変更が発生したときとは異なる順序で表示される場合があり、同じアイテムの通知がストリームに複数回表示される場合があります。アイテムの変更が発生した順序を識別し、重複したレコードを識別するために、ApproximateCreationDateTime 属性を確認することができます。

Lambda がストリーム上で処理を実行する方法

イベントソースマッピングは、ストリームおよびキューに基づくサービスからアイテムを読み取り、レコードのバッチで関数を呼び出す Lambda リソースです。イベントソースマッピングを使用して、 Lambda 関数を直接呼び出さないサービスのストリームまたはキューからのイベントを処理できます。 Lambda は DynamoDB Streams から直接呼び出されるのではなく、そのリソースを介して呼び出されることを理解して、各ソリューションと問題の特性に再度アプローチしましょう。
イベントソースマッピング構成の MaximumRecordAgeInSeconds および MaximumRetryAttempts 値を設定することで、基本的な再試行処理を処理できます。ただし、 Lambda コードのバグやデプロイ時の問題など、再試行のみでは解決できない障害が発生する可能性があります。
イベントソースマッピングリソースを参照すると、処理できなかったレコードに関する通知を Amazon Simple Queue Service （Amazon SQS）キューまたは Amazon Simple Notification Service （Amazon SNS）トピックに転送するための On-failure destination 設定を構成できます。次の例は、 DynamoDB ストリームの呼び出しレコードを示しています。

{ 
    "requestContext": { 
        "requestId": "316aa6d0-8154-xmpl-9af7-85d5f4a6bc81", 
        "functionArn": "arn:aws:lambda:us-east-2:123456789012:function:myfunction", "condition": "RetryAttemptsExhausted", 
        "approximateInvokeCount": 1 
    }, 
    "responseContext": { 
        "statusCode": 200, 
         "executedVersion": "$LATEST", 
        "functionError": "Unhandled" 
    }, 
    "version": "1.0", 
    "timestamp": "2019-11-14T00:13:49.717Z", 
    "DDBStreamBatchInfo": { 
        "shardId": "shardId-00000001573689847184-864758bb", 
        "startSequenceNumber": "800000000003126276362", 
        "endSequenceNumber": "800000000003126276362", 
        "approximateArrivalOfFirstRecord": "2019-11-14T00:13:19Z", "approximateArrivalOfLastRecord": "2019-11-14T00:13:19Z", 
        "batchSize": 1, 
        "streamArn": "arn:aws:dynamodb:us-east-2:123456789012:table/mytable/stream/2019-11-14T00:04:06.388" 
    } 
}

前述の情報に基づき、 DynamoDB Streams からレコードを取得して再試行する必要があります。DynamoDB Streams からレコードを取得して処理する際、すべてのイベントが時系列順に配信されるわけではないため、イベントの順序が乱れる可能性があります。
BatchSize が 1 を超え、一時的な障害により一部のアイテムの処理に失敗した場合、 Lambda は以前に処理されたレコードを含むバッチ全体を再試行します。適切に処理しないと、これにより重複イベントが発生する可能性があります。
さらに、イベントソースマッピングの開始位置が LATEST に設定されている場合、一部のイベントが見逃される可能性があります。
イベントソースマッピングに設定された MaximumRetryAttempts や MaximumRecordAgeInSeconds のような値は、初期設定と異なり、エラー処理や状況に応じて変更する必要があることがあります。この場合、一部のレコードが意図せず見逃される可能性があります。
この問題を解決するために開始位置を TRIM_HORIZON に変更すると、 DynamoDB Streams 内のすべてのデータが最初からイベントコンシューマーに配信され、イベントの重複処理が発生する可能性があります。

DynamoDB Streams と Kinesis Data Streams による問題解決

DynamoDB Streams と Kinesis Data Streams が同様の問題を解決する能力があると信じています。この記事では、次のユースケースについて説明します：

すべてのストリーム処理関数を冪等に書くことは可能でしょうか？
Lambda 関数で問題が発生したときに再試行することはできますか？

冪等性

ストリーム処理において最も重要なことの 1 つは、ロジックを冪等に設計することです。受信したイベントを検証し、以前に処理されたかどうかを判断する必要があります。イベントコンシューマーを冪等に書くことで、多くの問題を解決することができます。
例えば、ストリーム内のイベントが乱れた順序で表示される状況を見てみましょう。

前述の図に示すように、イベントの処理順序が不適切なためにデータ整合性が失われる可能性があります。
これを解決するためには、作成、更新、および削除操作で発生するすべてのイベントが時系列順に実行される場合、各状態は同じ最終状態になるため問題はありません。
言い換えれば、現在の最終状態が最新のイベントの結果である場合、前述の問題は簡単に解決できます。
これを実行するために、時刻を表すタイムスタンプが大きい場合にのみ更新が実行されるように実装を再設計しましょう。これにより、現在の状態以降のイベントのみが実行されます。

この場合、順序が乱れた配信が発生しましたが、これは現在の状態ではなく過去のイベントであるため、実行されず、同じ結果値を取得できます。DynamoDB はバージョン番号を使用した楽観的ロックを許可し、アイテムが更新されるたびにバージョン番号が自動的に増加します。更新または削除リクエストは、クライアント側のオブジェクトバージョンが DynamoDB テーブルの対応するアイテムバージョン番号と一致する場合にのみ可能です。この特性を利用すれば、問題を解決できます。
前と同じロジックを実行する場合、作成および更新操作には問題ありませんが、削除に関して問題になるケースがあります。

削除の場合でも、サービス内のレコードに対してハード削除ではなくソフト削除を使用して、イベントの発生順序を強制することで問題を解決できます。たとえば、A が削除され、その削除時刻に関する情報がある場合、その情報を使用してイベントを無視することができます。

障害発生時のリトライ戦略

すべてのロジックが冪等に書かれていることを前提に、問題が発生したときにリトライできるかどうかについて話しましょう。
Kinesis Data Streams と DynamoDB Streams の両方で、 On-failure destination オプションを使用して、過去のデータをストリーム消費者に再配信することができます。しかし、二つのストリームの戦略は異なる場合があります：

DynamoDB Streams – DynamoDB Streams は、イベントソースマッピングの開始位置に LATEST および TRIM_HORIZON を提供します。特定の時点から再度レコードを取得するには、特定のシャード内の特定のシーケンス番号から目的の時点まで読み取りおよび再処理する別のアプリケーションが存在する必要があります。
Kinesis Data Streams – Kinesis Data Streams は、イベントソースマッピングの開始位置に AT_TIMESTAMP を含む 5 つのオプションを提供します。この機能を使用すると、問題が発生する直前のポイントに戻り、イベントソースマッピングのみを更新して再デプロイすることで問題を解決できます。

チャネルコーポレーションの選択

DynamoDB が提供する 2 つのストリームを使用して他のサービスとデータを同期する際に発生する可能性のあるケースについて検討しました。運用上の考慮事項やコストの観点から、特定のストリームを無条件に使用する方が良いとは言い難いです。したがって、チャネルコーポレーションは特定の基準に基づいて両方のストリームを使用します。
次のユースケースでは、DynamoDB Streams を使用します。

イベントが時系列順に発生することが重要な場合
問題が発生したときにエラー回復コストが高くても良い場合

次のユースケースでは、 Kinesis Data Streams を使用します。

問題が発生したときに特定の時点から迅速に回復することが重要な場合
2 つ以上の Lambda 関数が同時にストリームを処理する必要がある場合

DynamoDB Streams を使用したオンラインスキーマ変更戦略

ストリームの使用のもう 1 つの例として、チャネルコーポレーションは DynamoDB Streams を使用してオンラインスキーマ変更を実行します。同じアプローチは異なる AWS アカウント間の移行にも使用できます。次の図は、ワークフローを示しています。

このワークフローには次のステップが含まれています。

最初のステップは二つのパートで構成されています
- DynamoDB に新しいスキーマを持つ新しいテーブルを作成します。
- 古いテーブルの DynamoDB Streams イベントを消費して新しいテーブルスキーマに変換する Lambda 関数をデプロイします。
Lambda 関数がデプロイされる前の履歴データを読み込み、新しいスキーマに適用します
新しい API サーバーをデプロイします

このプロセスにより、スキーマ変更が大幅に行われる場合でも、ライブスキーマ変更を実行できます。ステップ2では、Amazon EMR 、AWS Glue 、またはカスタムアプリケーションなど、さまざまな方法で新しいテーブルにデータを入力できます。
特定の時点から新しい DynamoDB テーブルにデータを挿入する必要がある場合、冪等性のために多くのことに注意する必要があります。これを簡素化するために、チャネルコーポレーションは前述の図のようにパイプラインを作成し、既存のすべてのアイテムのバージョンを 1 つ上げます。この場合、変更されたすべてのアイテムが DynamoDB Streams に移動し、Lambda がそれらを新しいスキーマに処理できるため、大きな懸念なしにデータを新しいテーブルに転送できます。

結論

DynamoDB を使用すると、スケーリングがほぼ無限であり、さまざまなダウンストリームサービスとの依存関係が排除されます。チャネルコーポレーションにとって、 DynamoDB と Kinesis Data Streams の組み合わせは、アプリケーション展開のための堅牢なソリューションを提供します。この組み合わせにより、デプロイメント中に問題が発生した場合でも、特定の時点から迅速に回復できます。その結果、開発者は信頼できるフォールバックメカニズムがあることを知って、いつでも自信を持ってデプロイメントを実行できます。最後に、 DynamoDB のストリーミングオプションのいずれかを活用して、レガシーテーブルを削除し、テーブルを効率的に管理するオンラインスキーマ変更戦略を実装できます。
同様のソリューションを自分のユースケースに実装することを検討し、質問があればコメントセクションに残してください。

著者について

TaeHun (Clsan) Yoon はコンピュータエンジニアリングの分野で経験豊富なプロフェッショナルである Channel Corp の最高技術責任者（ CTO ）として革新的なソリューションを牽引しています。チャット体験の向上に注力し、 TaeHun と彼のチームは顧客が直面するさまざまな課題の解決に尽力しています。

Changsoon (CS) Kim は、 Channel Corp の DevOps エンジニアです。開発と運用の間の問題を効率的に解決することに関心を持っています。

Sungbae Park は、 AWS スタートアップチームのアカウントマネージャーであり、 AWS SaaS TFC （Technical Field Communities）のレギュラーメンバーです。現在、 B2B ソフトウェアスタートアップが AWS で成長し成功するのを支援することに注力しています。以前は、 MSP 、 SI 、 ISV パートナーと共に相互成長を促進するパートナーデベロップメントマネージャーとして働いていました。

Hyuk Lee は、韓国に拠点を置くシニア DynamoDB スペシャリストソリューションアーキテクトです。彼は Amazon DynamoDB の機能を使用して顧客のアーキテクチャをモダナイゼーションするのを支援することが大好きです。

Amazon Web Services ブログ