data-aws

データ AWS スキル

バージョン： 1.1 スタック： AWS（Glue、Lambda、S3、Redshift、Athena、Step Functions）

AWS のデータサービスは使用量に応じて課金されます。クエリを実行するたびにデータがスキャンされ、DPU 時間ごとに料金が発生し、パーティション分割されていないテーブルでは $5/TB のフルスキャンが発生します。適切に設計されたデータレイクと構造の不適切なデータレイクの違いは、パフォーマンスだけではなく、実際のコストにも表れます。同じ結果を得る場合でも、パーティション分割されていない Athena クエリは、パーティション分割されたクエリの 100 倍のコストがかかることがあります。デフォルトの DPU 数で Glue ジョブを実行すると、コンピューティングリソースが無駄になります。ハンドラー内でクライアントを初期化する Lambda 関数は、呼び出しのたびにコールドスタート時間を浪費します。

こうしたパターンは、単なるベストプラクティスではなく、コスト管理策です。

---

対象範囲と境界

このスキルの対象：

• S3 データレイクのパターン
• Glue ETL ジョブとクローラー
• データ処理用の Lambda
• Athena クエリのパターン
• Redshift の最適化
• Step Functions によるオーケストレーション

他のスキルの対象：

• data-iac：インフラストラクチャ向けの Terraform/CDK
• data-pipelines：一般的な ETL パターン
• security：IAM、暗号化、シークレット

このスキルを使用する場面： AWS 上でデータソリューションを構築するとき。

---

基本原則

1. 基盤としての S3 — まずデータレイクを構築し、提供用データにはデータウェアハウスを使用します。
2. すべてをパーティション分割 — S3 のパスとテーブルは、少なくとも日付でパーティション分割します。
3. 適切なサイズのコンピューティング — 小規模（<10 GB、<15 分）には Lambda、大規模には Glue、提供用には Redshift を使用します。
4. コストを意識する — クエリを実行するたびにデータがスキャンされ、DPU ごとに料金が発生します。パーティション分割、圧縮、列指向フォーマットを活用します。
5. 可能な限りイベント駆動にする — ポーリングではなく、S3 イベントで処理をトリガーします。
6. スキーマを明示する — Glue Catalog でスキーマを定義し、本番環境ではクローラーに依存しないでください。

---

サービスの選択

データ量	頻度	必要なレイテンシー	使用するサービス
<10 GB	イベント駆動	秒単位	Lambda
>10 GB	スケジュール実行	分単位	Glue
任意	アドホック SQL	秒単位	Athena
大規模	同時 BI ユーザー	1 秒未満	Redshift
ストリーミング	継続的	ミリ秒単位	Kinesis

詳しい比較とコストの内訳については、references/service-selection.md を参照してください。

---

S3 データレイク

パス構造

s3://{project}-{env}-data/
├── raw/                                    # Immutable source data
│   └── source={source}/year={YYYY}/month={MM}/day={DD}/
├── staged/                                 # Cleaned, validated
│   └── table={table}/year={YYYY}/month={MM}/day={DD}/
├── curated/                                # Business logic applied
│   └── table={table}/year={YYYY}/month={MM}/day={DD}/
└── aggregated/                             # Pre-computed metrics
    └── table={table}/year={YYYY}/month={MM}/day={DD}/

ルール：

• Athena がパーティションを自動検出できるよう、Hive 形式のパーティション（key=value/）を使用します
• ステージング以降では Parquet 形式を使用します（列指向、圧縮済み、分割可能）
• Raw データは元の形式（JSON、CSV）のまま保持し、ソースデータは決して変更しません
• パーティションあたり最低 1 ファイルとし、数百万個の小さなファイルは避けます（各ファイル 128 MB～1 GB を目安にします）

---

Athena

パーティションプロジェクション

MSCK REPAIR TABLE とパーティション管理のオーバーヘッドをなくします。

CREATE EXTERNAL TABLE events (
    event_id STRING,
    user_id STRING,
    event_type STRING,
    payload STRING
)
PARTITIONED BY (year STRING, month STRING, day STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS PARQUET
LOCATION 's3://bucket/curated/table=events/'
TBLPROPERTIES (
    'projection.enabled' = 'true',
    'projection.year.type' = 'integer',
    'projection.year.range' = '2020,2030',
    'projection.month.type' = 'integer',
    'projection.month.range' = '1,12',
    'projection.month.digits' = '2',
    'projection.day.type' = 'integer',
    'projection.day.range' = '1,31',
    'projection.day.digits' = '2',
    'storage.location.template' =
        's3://bucket/curated/table=events/year=${year}/month=${month}/day=${day}'
);

クエリコストの管理

Athena はスキャンしたデータ 1 TB あたり $5 を課金します。次の方法でコストを削減します。

• パーティション列で必ずフィルタリングします（WHERE year = '2026' AND month = '02'）
• 列指向フォーマットを使用します（Parquet では要求された列だけがスキャンされます）
• データの確認中は LIMIT を使用します
• ワークグループでクエリのデータスキャン上限を設定します

---

Glue ETL

PySpark ジョブのパターン

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.context import SparkContext

args = getResolvedOptions(sys.argv, ["JOB_NAME", "SOURCE_DATABASE", "SOURCE_TABLE", "OUTPUT_PATH"])

sc = SparkContext()
glue_context = GlueContext(sc)
spark = glue_context.spark_session
job = Job(glue_context)
job.init(args["JOB_NAME"], args)

# Read from Glue Catalog
source_frame = glue_context.create_dynamic_frame.from_catalog(
    database=args["SOURCE_DATABASE"],
    table_name=args["SOURCE_TABLE"],
    push_down_predicate="year='2026' AND month='02'"  # Partition pruning
)

# Transform
mapped_frame = ApplyMapping.apply(
    frame=source_frame,
    mappings=[
        ("event_id", "string", "event_id", "string"),
        ("user_id", "string", "user_id", "string"),
        ("event_type", "string", "event_type", "string"),
        ("timestamp", "string", "event_ts", "timestamp"),
    ]
)

filtered_frame = Filter.apply(
    frame=mapped_frame,
    f=lambda row: row["event_type"] is not None
)

# Write as Parquet, partitioned
glue_context.write_dynamic_frame.from_options(
    frame=filtered_frame,
    connection_type="s3",
    connection_options={
        "path": args["OUTPUT_PATH"],
        "partitionKeys": ["year", "month", "day"]
    },
    format="parquet",
    format_options={"compression": "snappy"}
)

job.commit()

Glue のルール

• 小さく始める：2 DPU から始め、ジョブが遅い場合にのみ増やします。DPU 1 つあたり $0.44/時間かかります。
• ジョブブックマーク：増分処理で有効にし、古いデータの再処理を避けます。
• 本番環境でクローラーを使用しない：スキーマの変更は下流の処理を壊します。Glue Catalog でスキーマを明示的に定義します。
• プッシュダウン述語：すべてのデータをロードした後ではなく、読み込み時にフィルタリングします。

---

Lambda

S3 イベント処理のパターン

import json
import logging
import boto3
from botocore.exceptions import ClientError

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Initialize outside handler for connection reuse across invocations
s3 = boto3.client("s3")

def handler(event, context):
    """Process S3 event: read from raw, transform, write to staged."""
    processed = 0
    errors = 0

    for record in event["Records"]:
        bucket = record["s3"]["bucket"]["name"]
        key = record["s3"]["object"]["key"]

        try:
            response = s3.get_object(Bucket=bucket, Key=key)
            data = json.loads(response["Body"].read())

            result = transform(data)

            output_key = key.replace("raw/", "staged/")
            s3.put_object(
                Bucket=bucket,
                Key=output_key,
                Body=json.dumps(result),
                ContentType="application/json"
            )
            processed += 1

        except ClientError as e:
            logger.error("S3 error processing %s: %s", key, e.response["Error"]["Code"])
            errors += 1
        except (json.JSONDecodeError, KeyError) as e:
            logger.error("Data error processing %s: %s", key, e)
            errors += 1

    logger.info("Processed %d records, %d errors", processed, errors)
    return {"processed": processed, "errors": errors}

Lambda のルール

• ハンドラーの外でクライアントを作成する：boto3.client() の呼び出しによって、ウォーム状態での呼び出し間で TCP 接続が再利用されます。
• メモリ = CPU：メモリを増やすと、CPU の割り当ても増えます。プロファイリングを行い、コスト効率が最適な設定を見つけます。
• 冪等な操作：S3 イベントは重複して配信されることがあります。同じ出力を 2 回書き込んでも安全な設計にします。
• デッドレターキュー：失敗したイベントを再処理できるよう、SQS DLQ を設定します。

---

Redshift

テーブル設計

CREATE TABLE fact_orders (
    order_id        BIGINT        NOT NULL ENCODE az64,
    customer_id     BIGINT        NOT NULL ENCODE az64,
    order_date      DATE          NOT NULL ENCODE az64,
    product_id      BIGINT        NOT NULL ENCODE az64,
    quantity         INTEGER       NOT NULL ENCODE az64,
    total_amount    DECIMAL(12,2) NOT NULL ENCODE az64,
    region          VARCHAR(50)   ENCODE zstd
)
DISTSTYLE KEY
DISTKEY (customer_id)     -- Collocates joins on customer_id
SORTKEY (order_date);     -- Optimizes date range filters

主な設計上の決定事項：

• DISTKEY：JOIN 条件で最もよく使用する列です。一致する行が同じノードに配置されます。
• SORTKEY：WHERE／ORDER BY で最もよく使用する列です。ゾーンマップによるプルーニングが可能になります。
• DISTSTYLE EVEN：明確な結合列がない場合に使用します。行を均等に分散します。
• DISTSTYLE ALL：小さなディメンションテーブル（1M 行未満）に使用します。すべてのノードにコピーされます。

COPY によるデータのロード

COPY fact_orders
FROM 's3://bucket/curated/table=orders/'
IAM_ROLE 'arn:aws:iam::123456789012:role/redshift-s3-read'
FORMAT AS PARQUET;

一括ロードには必ず INSERT ではなく COPY を使用してください。すべてのノードで並列に S3 から直接読み取ります。

メンテナンス

-- Run after significant data changes
VACUUM FULL fact_orders;    -- Reclaims space, re-sorts rows
ANALYZE fact_orders;        -- Updates query planner statistics

---

Step Functions

ETL ワークフローパターン

{
  "StartAt": "RunGlueJob",
  "States": {
    "RunGlueJob": {
      "Type": "Task",
      "Resource": "arn:aws:states:::glue:startJobRun.sync",
      "Parameters": {
        "JobName": "etl-raw-to-curated",
        "Arguments": {
          "--SOURCE_DATABASE": "raw_db",
          "--SOURCE_TABLE": "events",
          "--OUTPUT_PATH": "s3://bucket/curated/table=events/"
        }
      },
      "Retry": [
        {
          "ErrorEquals": ["Glue.AWSGlueException"],
          "IntervalSeconds": 60,
          "MaxAttempts": 2,
          "BackoffRate": 2.0
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "NotifyFailure"
        }
      ],
      "Next": "RefreshAthenaPartitions"
    },
    "RefreshAthenaPartitions": {
      "Type": "Task",
      "Resource": "arn:aws:states:::athena:startQueryExecution.sync",
      "Parameters": {
        "QueryString": "MSCK REPAIR TABLE curated_db.events",
        "WorkGroup": "data-team"
      },
      "Next": "NotifySuccess"
    },
    "NotifySuccess": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:us-east-1:123456789012:etl-alerts",
        "Message": "ETL pipeline completed successfully"
      },
      "End": true
    },
    "NotifyFailure": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:us-east-1:123456789012:etl-alerts",
        "Message.$": "States.Format('ETL pipeline failed: {}', $.Error)"
      },
      "End": true
    }
  }
}

ルール：

• .sync サフィックス：ジョブの完了を待機します（指定しない場合、Step Functions はジョブを開始するだけで次に進みます）。
• Catch の前に Retry：エラー処理に移る前に、一時的な障害から自動復旧できるようにします。
• すべての状態にタイムアウトを設定：実行が停止したまま無期限に続くのを防ぎます。

---

アンチパターン

アンチパターン	問題	対策
パーティション分割されていない S3／テーブル	全件スキャンが発生し、処理が遅くコストも高くなります	最低でも日付でパーティション分割します
数百万個の小さなファイル	一覧表示が遅く、リクエストコストが高くなります	128 MB～1 GB のファイルにまとめます
大きなファイルに Lambda を使用	タイムアウトやメモリ制限が発生します	10 GB を超える場合は Glue または Fargate を使用します
本番環境で Glue クローラーを使用	スキーマが変動し、予測できなくなります	スキーマを明示的に定義します
ソートキー／分散キーを設定せずに Redshift を使用	クエリのパフォーマンスが低下し、データの再配置が発生します	クエリパターンに基づいてキーを定義します
パーティションプルーニングを使わずに Athena を使用	1 TB のスキャンごとに $5 がかかり、データセット全体がスキャンされます	必ずパーティション列で絞り込みます
新しいファイルを S3 でポーリング	無駄が多く、遅延が発生します	S3 イベント通知または EventBridge を使用します
ARN／パスをハードコード	環境間で動作しなくなります	ジョブ引数または SSM でパラメーター化します

---

コストの目安

サービス	コスト要因	コストを抑える方法
S3	ストレージ＋リクエスト	圧縮する、Glacier へのライフサイクル移行を設定する、小さなファイルを避ける
Athena	スキャン 1 TB あたり $5	パーティション分割する、Parquet を使う、必要な列だけを選択する
Glue	DPU 時間あたり $0.44	2 DPU から開始する、プッシュダウン述語を使用する
Lambda	実行時間 × メモリ	メモリを適切なサイズにする、接続を再利用する
Redshift	RPU 時間またはノード時間	アイドル時に一時停止する、負荷が変動する場合は Serverless を使用する

---

チェックリスト

• S3 パスで Hive 形式のパーティション（最低でも year/month/day）を使用している
• ステージング層以降のデータ形式が Parquet である
• ファイルサイズが 128 MB～1 GB である（小さなファイルの問題がない）
• Athena テーブルでパーティションプロジェクションまたは管理対象パーティションを使用している
• Lambda 関数の実行時間が 15 分未満、データ量が 10 GB 未満で、DLQ が設定されている
• Glue ジョブのスキーマが明示的に定義されている（本番環境でクローラーを使用していない）
• Glue ジョブが最小の DPU 数で開始し、ブックマークが有効になっている
• Redshift テーブルのソートキーと分散キーが定義されている
• Step Functions のすべての状態に Retry、Catch、タイムアウトが設定されている
• CloudWatch でコストアラートが設定されている

---

参考資料

• references/service-selection.md — AWS の各種データサービスを使い分ける方法とコスト比較

アセット

• assets/aws-data-checklist.md — AWS データインフラストラクチャの完全なチェックリスト（S3、Athena、Glue、Lambda、Redshift、Step Functions、モニタリング、セキュリティ）