【ITニュース解説】Building a Near Real-Time Stock Analytics Pipeline on AWS (and the Silent Bug That Hid My Data)

このプロジェクトは、AWS上にほぼリアルタイムの株価データ分析パイプラインを構築し、データの流れ、サーバーレス処理、ストレージ、分析、通知がどのように連携するかを学ぶことを目的とした。外部APIからデータを取り込み、クラウド環境で処理、保存、分析し、最終的にトレンドに基づいてメールで通知するシステムである。

パイプラインのアーキテクチャは、まず外部の株価APIからPythonスクリプトで株価データを取得するところから始まる。このPythonスクリプト（プロデューサー）は、取得した株価レコードをAmazon Kinesis Data Streamsというストリーミングサービスに送信する。Kinesisは大量のデータをリアルタイムで収集・処理するためのサービスだ。次に、Kinesisに届いたレコードは、AWS Lambdaというサーバーレスなコンピューティングサービスによって処理される。Lambda関数は、設定されたトリガーに基づいて自動的に実行されるため、サーバーを管理する必要がない。この処理用Lambda関数は、株価レコードを分析し、加工済みのデータをAmazon DynamoDBに、そして生データをAmazon S3に保存する。DynamoDBは高速なキーバリュー型データベースで、素早いデータ参照に適している。一方、S3はオブジェクトストレージサービスであり、大量の生データを安価に長期保存するのに使われる。

保存されたデータは、さらに分析のために活用される。S3に保存された生データは、AWS Glueというデータカタログサービスによって構造化され、Amazon Athenaを使ってSQLクエリで分析できるようになる。AthenaはS3上のデータを直接クエリできるサービスで、データウェアハウスを構築・維持する手間を省ける。また、DynamoDBに保存されたデータに変化があった場合、DynamoDB Streamsがその変更を検知し、別のLambda関数（アラート用Lambda）を起動する。このアラート用Lambda関数は、株価のトレンドを検出し、Amazon SNS（Simple Notification Service）を介してメールで通知を送信する。SNSはメッセージを複数の購読者に一斉送信できるサービスだ。システム全体の監視やデバッグにはAmazon CloudWatchが利用され、ログの収集やメトリクスの監視を行う。

プロジェクトのスターターコードをベースに、いくつかの重要な変更を加えた。まず、設定値をコード内に直接記述するのではなく、環境変数や.envファイルから読み込むように変更した。これにより、機密情報がコードに混入するのを防ぎ、設定の変更を容易にした。AWSの認証情報もリポジトリに含めず、AWS CLIの設定を通じて取得するようにした。さらに、プロデューサーのIAM権限をKinesisへの書き込みのみに制限し、最小権限の原則に従った。ログの出力も必要な情報だけに絞り、可読性を高め、情報漏洩のリスクを減らした。GitHubの秘密情報保護機能も活用し、誤って認証情報などをコミットしないよう対策を講じた。

このプロジェクトで最も貴重な経験となったのは、データの流れが止まる沈黙のバグに直面し、それをデバッグした過程である。Pythonスクリプトは株価データをKinesisに送信したと成功メッセージを表示していたにもかかわらず、DynamoDBにはデータがなく、Lambdaのログも確認できなかった。まず、パイプラインの中間地点である処理用Lambda関数を単独でテストし、正しく動作することを確認した。次に、LambdaとKinesisを接続するイベントソースマッピングを確認したが、ここにも問題はなかった。これにより、問題がプロデューサー側、つまりKinesisへのデータ送信にある可能性が高いと判断した。プロデューサーの実行環境を詳しく調べると、Pythonのdotenvパッケージがインストールされておらず、.envファイルが読み込まれていなかったことが判明した。さらに悪いことに、このエラーがtry-exceptブロックでサイレントに無視されていたため、エラーが表示されなかったのだ。結果として、プロデューサーは意図しないAWSリージョンにデータを送信していた。この問題は、必要なパッケージをインストールし、設定を修正することで解決した。この経験から、プロデューサーが成功メッセージを出しても、データが意図したコンシューマーに届いているとは限らないという重要な教訓を得た。

技術的な選択にはいくつかの考慮があった。生データをS3に、加工済みデータをDynamoDBに保存したのは、S3が安価な長期保存に適しているのに対し、DynamoDBは高速なデータ参照に向いているためである。S3の生データは、将来の分析や再処理の可能性を残すためにも重要だった。分析には、専用のデータベースを維持する代わりに、AWS GlueとAthenaを活用した。これにより、サーバーを運用することなくS3上のデータをSQLでクエリでき、コスト効率も向上した。また、DynamoDBに数値を保存する際には、Pythonのfloat型ではなくDecimal型を使用した。これは、float型がboto3（PythonからAWSサービスを操作するSDK）のDynamoDBシリアライザでサポートされておらず、浮動小数点の精度問題を避けるためでもある。Kinesisはオンデマンドモードを選択し、事前にシャーディング容量を予測・設定する手間を省いた。小バッチ処理は、パイプラインのシンプルさを保ちつつ、バッチ処理がレイテンシーと呼び出しコストに与える影響を学ぶために採用した。

データに関する学びも多かった。例えば、30秒ごとに株価データを送信していても、市場が閉まっている期間やデータの更新がない期間では、同じ株価が繰り返し送られることに気づいた。これは、データの取り込み頻度とデータの鮮度は別物であるという重要な区別を示している。また、トレンド検出のアラート機能は、実際の市場のように価格が大きく変動するデータでテストしないと、その効果を検証できないことを学んだ。タイムスタンプの形式も重要で、異なる形式で保存されたタイムスタンプを文字列として比較すると、予期せぬ結果を招く可能性があるため、一貫したUTC形式を使用し、日時オブジェクトとして比較するべきだと理解した。

プロジェクトの完了後には、不必要な継続課金を避けるため、構築したAWSリソースをすべて削除するクリーンアップ作業を行った。Kinesis、Lambda、DynamoDB、S3、Glue、CloudWatchロググループ、IAMロールなど、関連するすべてのリソースを対象とした。アプリケーションを停止しただけではクラウド上のリソースは自動的に削除されないため、手動での確認と削除が不可欠である。

今後の改善点としては、より頻繁に更新される市場データを利用して、トレンド検出ロジックをより現実的な状況でテストすること、アラート機能のタイムウィンドウロジックを修正し、テストケースを作成することが挙げられる。また、Lambda関数のIAM権限を最小限に抑え、設定値の外部化をさらに進めること、そしてTerraformのようなInfrastructure as Codeツールを導入してインフラの構築と削除を自動化・再現可能にすることも重要だ。さらに、リトライメカニズムやデッドレターキューの導入など、障害発生時の処理を強化する必要がある。

このプロジェクトを通じて、単にAWSサービスを接続する方法だけでなく、分散システムにおいて一部のコンポーネントが正常に見えても、全体として意図した動作をしない場合に、どのように問題を探し、解決するかを学ぶことができた。データのエンドツーエンドでの追跡、設定エラーの可視化、コンポーネントの単体テストの重要性、そしてインフラだけでなくデータそのものの特性を理解することの価値を痛感した。セキュリティとコストを常に意識してシステムを構築し、不要なリソースは適切にクリーンアップすることの重要性も再認識した。この経験は、ストリーミング、サーバーレス処理、ストレージ、分析、イベント駆動型通知といった様々なクラウドサービスの連携に対する実践的な理解を深めるのに役立った。