data-aws

Data AWS Skill

版本： 1.1 技术栈： AWS（Glue、Lambda、S3、Redshift、Athena、Step Functions）

AWS 数据服务按用量收费——每次查询都会扫描数据，每个 DPU 小时都会产生费用，每张未分区表都会导致全表扫描，费用为每 TB 5 美元。架构良好的数据湖与结构糟糕的数据湖之间的差别不只是性能——还是真金白银。对于相同的查询结果，未分区的 Athena 查询费用可能是分区查询的 100 倍。Glue 作业使用默认 DPU 数量会浪费计算资源。在处理程序内部初始化客户端的 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 用户	亚秒级	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}/

规则：

• 使用 Hive 风格分区（key=value/），以便 Athena 自动发现分区
• 暂存层及之后的层级使用 Parquet 格式（列式存储、压缩、可拆分）
• 原始层保留原始格式（JSON、CSV）——绝不修改源数据
• 每个分区至少包含一个文件，避免产生数百万个小文件（每个文件以 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 每小时收费 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 事件可能重复投递。重复写入相同输出应当安全。
• 死信队列：配置 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：没有明确的 JOIN 列时使用。将行均匀分布。
• 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;

批量加载时始终优先使用 COPY，而不是 INSERT——它会从 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 只会启动作业，然后继续执行）。
• 先 Retry，后 Catch：让瞬时故障先自动恢复，再进入错误处理。
• 为每个状态设置超时：防止执行卡住并无限期运行。

---

反模式

反模式	问题	解决方法
未分区的 S3/表	全表扫描，速度慢且成本高	至少按日期分区
数百万个小文件	列表操作慢，请求成本高	合并为 128 MB–1 GB 的文件
使用 Lambda 处理大型文件	超时、内存限制	对于 >10 GB 的文件，使用 Glue 或 Fargate
在生产环境中使用 Glue 爬网程序	Schema 漂移、结果不可预测	显式定义 Schema
Redshift 未设置排序键/分配键	查询性能差、需要重新分布数据	根据查询模式定义键
Athena 未进行分区裁剪	以 $5/TB 的价格扫描整个数据集	始终筛选分区列
轮询 S3 检查新文件	浪费资源、延迟高	使用 S3 事件通知或 EventBridge
硬编码 ARN/路径	在不同环境中无法正常工作	使用作业参数或 SSM 参数化

---

成本经验法则

服务	成本因素	节省成本的方法
S3	存储 + 请求	压缩，将数据生命周期转至 Glacier，避免使用小文件
Athena	$5/TB 扫描量	分区、使用 Parquet、仅选择所需列
Glue	$0.44/DPU 小时	从 2 个 DPU 开始，使用下推谓词
Lambda	持续时间 x 内存	合理配置内存，复用连接
Redshift	RPU 小时或节点小时	空闲时暂停，负载变化时使用 Serverless

---

检查清单

• S3 路径使用 Hive 风格分区（至少按年/月/日分区）
• 从暂存层开始，数据均采用 Parquet 格式
• 文件大小在 128 MB 到 1 GB 之间（避免小文件问题）
• Athena 表使用分区投影或托管分区
• Lambda 函数运行时间 <15 min、处理数据量 <10 GB，并配置了 DLQ
• Glue 作业显式定义 Schema（生产环境不使用爬网程序）
• 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、监控、安全）