はじめに：スクリプトパイプラインの落とし穴
データパイプラインは、たいてい小さなスクリプトから始まります。CSVを読み込み、日付形式を変換し、特定のカラムをマッピングするだけの単純な処理です。しかし、時間が経つにつれて、似たような変換ロジックが複数のファイルに重複し、新しいデータセットを追加するたびに既存のスクリプトをコピー＆修正するようになります。
この方式の最大の問題は、ワークフローの意図（Intent）がコードの中に埋もれてしまうことです。後日「このパイプラインは正確に何をしているのか？」という質問に答えるには、コードを最初から解析しなければなりません。特に金融、製薬（臨床試験）など規制の厳しい環境では、これが致命的です。監査（Audit）の準備に数週間かかり、新しいデータセットを追加するたびに全体のデプロイサイクルを待つ必要があります。
この記事では、Specification-Driven Compositionパターンを紹介します。ワークフローの「何を」と「どのように」を分離し、より柔軟で管理しやすいデータパイプラインを構築する方法です。AWS Lambda、Step Functions、S3、OpenSearch Serviceを活用したサーバーレス実装例も併せて解説します。
コアアイデア：意図と実行の分離
Specification-Driven Compositionは3つのレイヤーで構成されます。
- Intent Layer（意図レイヤー） – ワークフローの振る舞いをJSON/YAMLスペックで定義
- Composition Layer（構成レイヤー） – スペックを検証し、実行可能なパイプラインにアセンブル
- Processing Layer（処理レイヤー） – 実際のデータ変換を実行する再利用可能な関数群
実際のスペック例（JSON）
{
"workflow": "monthly_finance_report",
"version": "1.0",
"source": {
"dataset": "raw_orders",
"format": "csv",
"location": "s3://data-lake/raw/orders/"
},
"target": {
"dataset": "standard_orders",
"format": "parquet",
"location": "s3://data-lake/standard/orders/"
},
"preprocessing": [
{
"capability": "standardize_columns",
"version": "2.1",
"params": {
"rename_map": {
"order_id": "order_identifier",
"order_total": "amount"
}
}
},
{
"capability": "cast_numeric",
"version": "1.0",
"params": {
"columns": ["amount", "tax", "shipping_cost"]
}
}
],
"mappings": [
{
"source_field": "order_date",
"target_field": "transaction_date",
"capability": "format_date",
"version": "1.2",
"params": {
"input_format": "%Y-%m-%d",
"output_format": "%Y%m%d"
}
},
{
"source_field": "amount",
"target_field": "amount_usd",
"capability": "normalize_currency",
"version": "3.0",
"params": {
"source_currency": "JPY",
"target_currency": "USD"
}
}
]
}
このスペックを見れば、「どのデータを取得し、どの前処理を行い、どのフィールドをどう変換して保存するか」がコードなしで明確にわかります。これが意図の可視化です。
AWSサーバーレス実装：Composer + Registry + Step Functions
アーキテクチャフロー
- ユーザーがS3バケットにJSONスペックをアップロード
- S3イベントがLambda Composer関数をトリガー
- Composerがスペックをパースし、OpenSearch Serviceに登録されたCapabilityレジストリから各変換関数のメタデータ（ARN、入出力スキーマ、権限）を照会
- Composerが検証を完了すると、Step Functionsステートマシン（ASL）を動的に生成して実行
- Step Functionsが各Capability Lambdaを順次呼び出し
- すべてのログはCloudWatch Logsに送信
Capabilityレジストリ設計の注意点
レジストリは単なるルックアップテーブルではなく、ガバナンス対象です。Capability定義はGitで管理し、CI/CDパイプラインでメタデータ検証とテストを通過した後にのみ新バージョンを登録する必要があります。
# 例：レジストリ照会関数（Python + boto3）
import boto3
from opensearchpy import OpenSearch, RequestsHttpConnection
def get_capability_metadata(capability_id, version):
"""
OpenSearchからCapabilityメタデータを照会します。
"""
client = OpenSearch(
hosts=[{'host': OPENSEARCH_ENDPOINT, 'port': 443}],
http_auth=('master_user', 'master_password'), # 実際はSecrets Managerを使用
use_ssl=True,
verify_certs=True,
connection_class=RequestsHttpConnection
)
query = {
"query": {
"bool": {
"must": [
{"term": {"capability_id": capability_id}},
{"term": {"version": version}}
]
}
}
}
response = client.search(index="capability-registry", body=query)
hits = response['hits']['hits']
if not hits:
raise ValueError(f"Capability {capability_id} v{version} not found")
return hits[0]['_source']
セキュリティ考慮事項
- S3バケットはSSE-KMS（カスタマーマネージドキー）で暗号化
- OpenSearchドメインはノード間暗号化およびHTTPS（TLS）を適用
- 各Capability Lambdaは必要なフィールドのみを受け取るようIAMポリシーを最小化
- 機密フィールドのタグ付け：スペックに
"sensitivity": "PHI"を明示し、Capabilityがこのタグをどう処理するか（維持/削除/追加）をレジストリに宣言 → Composerが自動でマスキングポリシーを生成
実務適用時の注意点と限界
このパターンはすべての状況に適しているわけではありません。 以下のケースでは、むしろ複雑さが増すだけです。
- 単発の変換処理：スペック作成とComposer設定のコストが、スクリプト1つ書くよりも大きい
- パイプラインが3～5本未満：重複排除の効果が薄い
- 超高速プロトタイピングが必要な場合：初期設定に時間がかかる
また、スペックスキーマの初期設計は慎重に行う必要があります。 柔軟にしすぎると検証ロジックが複雑になり、制限しすぎると多様なワークフローを表現できません。導入時は既存のパイプライン1本を選んでスペック化してみて、不足点を発見しながら段階的に改善することをおすすめします。
次のステップとしての学習方向
- AWS Lambdaイベント駆動アーキテクチャガイド – S3アップロードをComposerに接続する基本パターン
- Step FunctionsとLambda統合ガイド – 動的ステートマシン生成方法
- CloudWatchカスタムメトリクス – パイプライン性能監視
合わせて読みたい記事：
参考資料： AWS Architecture Blog - Specification-driven composition for flexible data workflows