📚 课程目录第3章数据摄取与传输集成

3. 数据摄取与传输集成

在 AWS 机器学习数据平台上，数据摄取层决定了后续特征工程、训练与推理能拿到什么质量、什么时效的数据。AWS 提供了多条摄取路径，它们并不是互相替代的关系，而是按“时效性—吞吐—数据源类型—成本”四个维度分工。理解每条路径的定位，才能为不同数据源选择正确的入口。

五条摄取路径的定位

Kinesis Data Streams（KDS）：低延迟、可重放的实时流。数据保留默认 24 小时，可延长至 365 天，支持多个消费者并行读取同一份数据，适合需要自定义消费逻辑（如 Flink、Lambda、KCL 应用）的场景。
Kinesis Data Firehose：全托管的“投递管道”。它不做复杂计算，只负责把流数据缓冲、转换、压缩后写入 S3、Redshift、OpenSearch 等目标，天然适合“落湖”这一动作。
Amazon MSK：托管 Kafka。当团队已有 Kafka 生态（Connect、Schema Registry、Kafka Streams）或需要 Kafka 协议兼容时使用，控制粒度比 KDS 更细。
AWS DMS：面向数据库的持续复制（CDC）。用于把 RDS、Aurora、Oracle、SQL Server 的增量变更以近乎实时的方式同步到 S3 或 KDS。
AWS Snowball：物理批量传输。当数据量达到 TB～PB 级、网络带宽成为瓶颈时，用设备离线搬运比走公网更快。

下面这张图把五条路径按“实时—批量”和“流式—数据库—文件”两个维度组织起来，并标出它们各自的典型下游目标，便于在选型时快速定位。

实时管道：KDS 与 Firehose 的配合

KDS 的核心抽象是“分片（Shard）”。每个分片提供 1 MB/s 写入和 2 MB/s 读取能力，分区键（Partition Key）决定记录进入哪个分片。如果分区键分布不均，会出现“热分片”，表现为某个分片写入被限流而其他分片空闲。实践中常用设备 ID、用户 ID 或带随机后缀的复合键来打散。

Firehose 通常接在 KDS 之后，也可以直接接收生产者数据。它的关键参数是缓冲：BufferSize（默认 5 MB）和 BufferInterval（默认 300 秒），两者满足其一即触发投递。对于训练数据落湖，建议把 BufferInterval 调到 60～120 秒，BufferSize 调到 64～128 MB，以生成更少、更大的 S3 对象，减少小文件问题。

# Firehose 投递到 S3 的关键配置示例
DeliveryStreamType: KinesisStreamAsSource
KinesisStreamSourceConfiguration:
  KinesisStreamARN: arn:aws:kinesis:us-east-1:123456789012:stream/ml-events
  RoleARN: arn:aws:iam::123456789012:role/firehose-kds-role
S3DestinationConfiguration:
  BucketARN: arn:aws:s3:::ml-data-lake
  Prefix: raw/events/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/
  BufferingHints:
    SizeInMBs: 128
    IntervalInSeconds: 120
  CompressionFormat: GZIP
  ErrorOutputPrefix: errors/!{firehose:error-output-type}/

注意 !{timestamp:...} 是 Firehose 的运行时表达式，会按记录到达时间生成 Hive 风格分区路径，这样 S3 上的数据可以直接被 Athena、Glue 或 Spark 按分区裁剪读取，避免全表扫描。

数据库增量：DMS 的 CDC 模式

当数据源是业务库时，全量快照只能解决首次同步，后续必须依赖 CDC（Change Data Capture）。DMS 通过读取数据库的日志（MySQL 的 binlog、PostgreSQL 的逻辑复制槽、Oracle 的 LogMiner）捕获 INSERT/UPDATE/DELETE，并以 op 字段标记操作类型写入目标。

一个常见的坑是：DMS 复制到 S3 时默认按 schema/table 目录写入，且每条记录是独立的 JSON 行。如果下游是 Glue 爬虫，需要开启 IncludeOpForFullLoad 并配置 TimestampColumnName，否则增量文件与全量文件结构不一致，爬虫会推断出错误的 Schema。建议在 DMS 任务中显式设置：

{
  "TargetMetadata": {
    "SupportLobs": true,
    "FullLobMode": false,
    "LimitedSizeLobMode": true,
    "LobMaxSize": 32
  },
  "FullLoadSettings": {
    "TargetTablePrepMode": "DO_NOTHING",
    "MaxFullLoadSubTasks": 8
  },
  "Logging": { "EnableLogging": true }
}

其中 TargetTablePrepMode: DO_NOTHING 表示 DMS 不删除目标表，适合持续增量场景；如果是首次全量，可改为 DROP_AND_CREATE。LOB 模式选择 LimitedSizeLobMode 并限制 32 KB，可以避免大字段拖慢复制吞吐。

批量搬运：Snowball 的适用边界

Snowball 不是“更快的网络”，而是“绕过网络”。判断是否该用 Snowball，可以用一个简单估算：假设待传数据 80 TB，专线带宽 1 Gbps，理论传输时间约为 80×1024×1024×8 / (1×10^9) ≈ 671 秒？这个数字明显不对——实际要考虑 TCP 开销、加密和磁盘 IO，有效吞吐通常只有标称的 60%～70%，且 1 Gbps 专线不可能全部给这一个任务。按 500 Mbps 有效带宽计算，80 TB 需要约 15 天。而 Snowball Edge 单台可装 80 TB，寄送往返约 5～7 天，明显更快。

使用 Snowball 时，数据在设备上会经过 KMS 加密，到达 AWS 后自动导入指定 S3 桶。导入完成后，S3 上会出现与本地目录结构一致的对象，此时再触发 Glue Crawler 或 Lambda 做后续处理。需要注意的是，Snowball 只解决“搬进来”，不解决“持续同步”，所以它通常与 DMS 或 KDS 组合使用：历史数据走 Snowball，增量数据走实时管道。

组合管道的搭建顺序

实际项目中，这五条路径往往组合成一条完整管道。推荐的搭建顺序是：

先确定 S3 数据湖的分区规范（如 year/month/day/hour），所有路径统一遵循。
历史数据用 Snowball 或 DMS 全量导入，形成基线数据集。
增量数据用 DMS CDC 或 KDS 接入，经 Firehose 落湖。
在 S3 上配置 Glue Crawler 定期更新元数据，供 Athena 和 SageMaker 使用。
对延迟敏感的特征，直接从 KDS 消费到 Lambda 或 Flink，写入在线特征存储。

选型速查：需要重放和多个消费者 → KDS；只需要落湖 → Firehose；已有 Kafka 生态 → MSK；数据库增量 → DMS；TB 级以上历史数据 → Snowball。

常见错误：把 Firehose 当作计算引擎做复杂聚合。Firehose 只支持 Lambda 做轻量转换（如字段裁剪、格式转换），复杂聚合应交给 KDS + Flink 或 Glue Streaming。

最后需要强调的是，无论走哪条路径，数据到达 S3 后都应立即具备可查询性。这意味着分区路径、文件格式（Parquet/ORC）、压缩方式（Snappy/GZIP）和元数据注册必须在一开始就统一。否则，后续每次训练都要额外写清洗脚本，摄取层的价值就大打折扣。