AWS S3を活用したデータ配信の裏側 ~ Speeda Datafeed事業のパイプラインとアクセス制御 ~

はじめに

こんにちは。JP Product Division に所属している datafeed チームです。

Datafeed チームでは、Speeda が保有する経済情報データを、API やクラウドストレージ連携を通じてお客様へ届けています。

今回はその中でも、AWS S3 を活用したクラウドストレージ連携にフォーカスし、安全なデータ配信を支えるパイプライン設計やアクセス制御の仕組みについてお話しします。

大量かつ重要な経済情報データを継続的に提供するには、データを生成するだけでなく、更新内容の確認、異常の検知、データ品質の検証、配信範囲の制御までを一連の仕組みとして設計することが重要です。

本記事では、S3 を中心に構成している配信パイプラインと、クロスアカウントでのアクセス制御について紹介します。

全体アーキテクチャ

まずは、データ生成からクライアント環境へ届くまでの全体像です。

[ 上流マイクロサービス群 ]
          │
          ▼
[ Api (Groovy / Micronaut) ]
          ▲
          │
[ Kubernetes CronJob ]
  毎日 0:00 に起動
          │
          ▼
[ AWS S3 : 作業用 tmp バケット ]
  ・データ減少検知
  ・オブジェクト属性を用いた差分判定
          │
          ▼
[ AWS S3 : 配信バケット ]
          │
          ├─► [ データ品質チェック Job ]
          │       ・Frictionless によるスキーマ・形式検証
          │       ・検証結果を監視・通知
          │
          ▼
[ クライアント AWS 環境 ]
  ・配信側 S3 バケットポリシー
  ・クライアント側 IAM Role
  ・AWS DataSync 等による同期

パイプラインのライフサイクルは、以下のように進みます。

定期実行

Kubernetes の CronJob が毎日 0:00 に起動し、Api を呼び出します。
差分比較と反映

tmp バケット上のデータに対して、データ件数の減少がないかを確認します。あわせて、S3 が提供するオブジェクト属性を利用して配信済みデータとの差分を確認し、更新が必要な場合のみ配信バケットへ反映します。
品質検証

配信バケットへの配置完了後、Python ライブラリの Frictionless を用いた検証 Job が実行されます。CSV のフォーマットやスキーマ、カラムごとのデータ品質を確認します。
クライアントによる取得

クライアント側の AWS 環境では、配信側の S3 バケットポリシーとクライアント側 IAM Role の権限設定に基づき、許可されたデータのみを取得します。

各ジョブの実行ライフサイクルは DB に永続化されています。ジョブの開始・終了時刻、実行結果、エラー情報などを記録しており、障害発生時の調査や再実行判断に利用しています。

処理中に例外が発生した場合は、AOP（インターセプター）を介して DB へエラーログが記録されるとともに、エンジニア向けの Slack チャンネルへ通知される仕組みになっています。

安全なデータ配信を支える2つのガードレール

作業用の tmp バケットから配信バケットへ反映する前段階で、主に2つのガードレールを設けています。

データの性質に応じた減少検知
S3 のオブジェクト属性を利用した差分判定

これにより、上流システムの異常や一時的な通信障害などによって、不完全なデータが配信されることを防ぎます。

1. データの性質に応じた減少検知

経済情報データには、日次で蓄積されるデータ、更新や訂正によって件数が変動するデータ、対象期間ごとに入れ替わるデータなど、さまざまな性質があります。

そのため、すべてのデータセットに一律の件数チェックを適用するのではなく、データ種別やジョブの特性に応じてフェイルセーフを設けています。

主なチェック例は以下のとおりです。

厳格な件数チェック

前回配信分と比較して、1件でもレコード数が減少していれば異常として扱うデータセットです。
しきい値ベースのチェック

前回配信分と比較して、一定割合以上の減少が発生した場合に異常として扱うデータセットです。

異常を検知した場合は、処理を中断して例外を発生させ、配信バケットへの反映をブロックします。

単純な件数比較だけでは、正しい更新まで止めてしまう可能性があります。一方で、何の検証も行わなければ、上流の一時障害で取得件数が大幅に減少したデータをそのまま配信してしまうリスクがあります。

そのため、データセットごとの更新特性を理解したうえで、過不足のないルールを設定することを重視しています。

2. S3のオブジェクト属性を利用した差分比較

生成したファイルの内容に変更がない場合、毎回配信バケットを更新する必要はありません。

すべてのオブジェクトをダウンロードして内容を比較すると、ネットワーク帯域や処理時間を余分に消費します。そこで、S3 が提供するオブジェクト属性を利用し、不要なデータ転送を抑えながら差分判定を行っています。

Datafeed のパイプラインでは、アップロード時にチェックサム情報を付与しています。差分比較時には GetObjectAttributes API を利用し、ETag やチェックサムなどのオブジェクト属性を取得して、対象データに応じた更新判定を行います。

tmp バケット上の新規データ
          │
          ▼
GetObjectAttributes で属性を取得
          │
          ▼
配信バケット上の既存データの属性と比較
          │
          ├─ 同一と判定された場合
          │      └─ 配信バケットへの反映をスキップ
          │
          └─ 更新が必要な場合
                 └─ 配信バケットへ反映

ETag やチェックサムは、データの生成・配信処理における差分検知の補助情報として利用しています。

データセットの特性やアップロード方式に応じて利用する属性を選び、同一と判定された場合は配信バケットへの反映をスキップします。更新が必要な場合のみ昇格処理を実行することで、不要な S3 操作やクライアント側の同期負荷を抑えています。

Frictionlessによる提供前後のデータ検証

Frictionless Framework（Python パッケージ名: frictionless）は、CSV・Excel・JSON・SQLなどの表形式データを記述・読み込み・検証・変換するためのオープンソースのPythonフレームワークです。

配信バケットへデータが配置された後、クライアントによる利用状況を継続的に監視しながら、Python のデータ検証ライブラリである Frictionless を用いたバリデーション Job を実行しています。

Frictionless を利用することで、CSV の検証ロジックをすべて自前で実装することなく、宣言的なスキーマ定義に基づいてフォーマットやデータ品質を確認できます。

主なチェック項目は以下のとおりです。

日付、数値、コード値など、カラムごとに定義した形式や値の妥当性
必須カラムの欠損や想定外のデータ型
不正な文字列や想定外の値の混入
データセットごとに定義した表記上のルール
スキーマ定義と実データのカラム構成の差異

バリデーションエラーを検出した場合は Slack へ通知が送られ、エンジニアが原因調査や再実行を速やかに行える状態を整えています。

現在は、配信後の検証結果を監視・通知する仕組みとして運用しています。今後は、検証済みデータのみをクライアントから参照可能にする公開フローへ段階的に改善していく予定です。

セキュアなクロスアカウントS3連携

クライアントへデータを提供するにあたり、パブリックなダウンロード URL を発行するのではなく、配信側の S3 バケットポリシーとクライアント側 IAM Role の権限設定を組み合わせたクロスアカウントのアクセス制御を採用しています。

インターネット上にデータを公開せず、AWS の認証・認可の仕組みを通じて、契約上許可されたクライアントだけが必要なデータへアクセスできる構成です。

アクセス制御の設計原則

項目	内容
公開範囲	インターネット公開なし。S3 Block Public Access を有効化
認可方式	配信側の S3 バケットポリシーと、クライアント側 IAM Role の権限設定
付与権限	用途に応じた読み取り権限。オブジェクト取得、対象プレフィックスの一覧取得、必要に応じたバケット情報取得
対象範囲	契約プランに応じたプレフィックス配下のみ
運用方針	クライアント・用途ごとに IAM Role を分離し、最小権限を維持

クロスアカウントで S3 にアクセスする場合、配信側のバケットポリシーだけでなく、クライアント側でも IAM Role に対して必要な権限を設定します。

これにより、どの AWS アカウントのどの IAM Role が、どの配信データへアクセスできるかを明示的に管理できます。

プレフィックスによるマルチテナント認可

単一の配信バケットを使いつつ、クライアントの契約プランに応じて、閲覧可能なプレフィックスを制限しています。

s3://<delivery-bucket-name>/
  ├── company_info/              # 企業情報
  ├── financial_statements/    # 財務情報
  ├── edinet/                  # 開示情報
  └── tdnet/                   # 適時開示情報

たとえば、あるクライアントが company_info と financial_statements を契約している場合、その IAM Role には該当プレフィックスに対する一覧取得とオブジェクト取得だけを許可します。

以下は、配信側アカウントに設定する S3 バケットポリシーの例です。

実際のアクセスでは、クライアント側でも対象 IAM Role に必要な IAM ポリシーを設定します。また、利用する同期方式や暗号化設定に応じて、追加の権限が必要になる場合があります。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowClientRoleToListBucket",
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::<Client-Account-ID>:role/<Client-IAM-Role>"
      },
      "Action": [
        "s3:ListBucket",
        "s3:GetBucketLocation"
      ],
      "Resource": "arn:aws:s3:::<delivery-bucket-name>",
      "Condition": {
        "StringLike": {
          "s3:prefix": [
            "company_info/*",
            "financial_statements/*"
          ]
        }
      }
    },
    {
      "Sid": "AllowClientRoleToReadObjects",
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::<Client-Account-ID>:role/<Client-IAM-Role>"
      },
      "Action": [
        "s3:GetObject",
        "s3:GetObjectTagging"
      ],
      "Resource": [
        "arn:aws:s3:::<delivery-bucket-name>/company_info/*",
        "arn:aws:s3:::<delivery-bucket-name>/financial_statements/*"
      ]
    }
  ]
}