4. 数据处理与转换集成

在AWS机器学习数据平台上，特征工程与ETL流水线很少由单一服务完成。真实的生产链路通常同时包含：对海量历史数据做分布式清洗与聚合、对增量文件做轻量解析与格式转换、对事件流做实时特征补齐，以及把上述步骤按依赖关系编排成可重试、可观测的工作流。AWS Glue、Amazon EMR、AWS Lambda 与 AWS Step Functions 正是覆盖这四类职责的核心组合。理解它们各自的“能力边界”和“交接方式”，比记住API名称更重要。

四个服务的职责分工

AWS Glue 是Serverless的ETL与数据目录服务。它的核心价值在于：Glue Crawler 自动推断S3数据的Schema并写入 Glue Data Catalog；Glue Job（基于Spark）执行分布式转换；Glue Studio 提供可视化编排。适合“数据量大、转换逻辑以SQL/Spark表达、无需长期维护集群”的场景。

Amazon EMR 是托管Hadoop/Spark/Hive/Presto集群。当转换逻辑复杂、需要精细控制Executor内存与并行度、需要复用已有Spark代码，或需要与HBase、Iceberg、Hudi深度集成时，EMR比Glue更合适。EMR on EKS 和 EMR Serverless 进一步降低了集群管理成本。

AWS Lambda 适合事件驱动的小数据量处理：S3对象创建触发、DynamoDB Streams触发、Kinesis记录级转换。单次执行最长15分钟、内存上限10GB，因此它不适合全量重算，但非常适合“文件到达即校验、小文件合并、格式转换、调用特征服务补全字段”。

AWS Step Functions 是状态机编排服务。它不处理数据，而是定义“谁先跑、谁依赖谁、失败后重试还是走补偿分支”。在特征工程中，它把 Glue Job、EMR Step、Lambda 函数、SageMaker Processing Job 串成一条有向无环图（DAG），并保留每次执行的输入输出。

协同模式一：Step Functions 编排 Glue Job 做批量特征回填

假设需要为推荐模型构建“用户近30天点击率”特征，数据在S3的原始日志区，每天新增约200GB Parquet文件。典型做法是：Step Functions 状态机先调用一个 Glue Crawler 更新分区元数据，再启动 Glue Job 执行聚合，最后调用 Lambda 校验输出行数是否在预期范围内。

Glue Job 的核心转换逻辑可以用 PySpark 表达。以下示例展示从原始事件表计算用户级特征并写入特征区：

import sys
from awsglue.context import GlueContext
from pyspark.context import SparkContext
from pyspark.sql import functions as F

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

# 从 Data Catalog 读取原始事件表
events = glueContext.create_dynamic_frame.from_catalog(
    database="ml_raw", table_name="user_events"
).toDF()

# 计算用户近30天点击率特征
features = (events
    .filter(F.col("event_date") >= F.date_sub(F.current_date(), 30))
    .groupBy("user_id")
    .agg(
        F.count("*").alias("total_events"),
        F.sum(F.when(F.col("event_type") == "click", 1).otherwise(0)).alias("click_count"),
        F.countDistinct("item_id").alias("distinct_items")
    )
    .withColumn("click_rate", F.col("click_count") / F.col("total_events"))
)

# 写入特征存储区，按日期分区
features.write.mode("overwrite").partitionBy("dt").parquet(
    "s3://ml-feature-store/user_click_features/"
)

Step Functions 中对应的状态定义（简化）如下，关键点是 Retry 与 Catch 的配置：

{
  "StartAt": "RunGlueJob",
  "States": {
    "RunGlueJob": {
      "Type": "Task",
      "Resource": "arn:aws:states:::glue:startJobRun.sync",
      "Parameters": {
        "JobName": "user_click_features_job",
        "Arguments": {
          "--dt": "$.run_date"
        }
      },
      "Retry": [
        {
          "ErrorEquals": ["Glue.AWSGlueException"],
          "IntervalSeconds": 60,
          "MaxAttempts": 2,
          "BackoffRate": 2.0
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "NotifyFailure"
        }
      ],
      "Next": "ValidateOutput"
    },
    "ValidateOutput": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:validate_feature_count",
      "Next": "Success"
    },
    "NotifyFailure": {
      "Type": "Task",
      "Resource": "arn:aws:sns:...:ml-pipeline-alerts",
      "End": true
    },
    "Success": { "Type": "Succeed" }
  }
}

这里 startJobRun.sync 表示 Step Functions 会等待 Glue Job 完成后再进入下一个状态，而不是只触发就返回。这是编排批量ETL时最常用的集成模式。

协同模式二：Lambda 做实时特征补齐，EMR 做复杂窗口计算

当Kinesis数据流中的每条记录需要补全用户画像特征时，Lambda 是合适的选择。它可以在毫秒级内查询 DynamoDB 或 SageMaker Feature Store，把特征拼接到记录中再写回下游。但Lambda不适合做“过去90天滑动窗口聚合”这类需要扫描大量历史数据的计算。

这类计算应交给 EMR。例如计算“用户过去90天购买间隔的方差”，需要按用户分组、按时间排序、计算相邻购买时间差，再求方差。Spark SQL 的窗口函数可以高效完成：

from pyspark.sql import Window
from pyspark.sql import functions as F

w = Window.partitionBy("user_id").orderBy("purchase_ts")

purchase_intervals = (purchases
    .withColumn("prev_ts", F.lag("purchase_ts").over(w))
    .withColumn("interval_days",
        F.datediff("purchase_ts", "prev_ts"))
    .filter(F.col("interval_days").isNotNull())
)

variance_features = (purchase_intervals
    .groupBy("user_id")
    .agg(F.variance("interval_days").alias("purchase_interval_var"))
)

EMR 的优势在于可以指定集群规模、Executor内存和并行度，对于TB级数据，可以通过调整 spark.sql.shuffle.partitions 和 spark.executor.memory 来优化性能。而 Glue 虽然也基于Spark，但对底层资源的控制粒度较粗，适合标准ETL而非极致调优场景。

关键集成点与判断依据

在实际搭建流水线时，以下几个判断点决定了服务选型：

• 数据量级：单次处理小于1GB且逻辑简单，优先Lambda；1GB～数百GB且无需长期集群，优先Glue；数百GB以上或需要复杂调优，优先EMR。
• 触发方式：事件驱动（S3对象创建、DynamoDB变更）优先Lambda；定时批量优先Glue Trigger或Step Functions Schedule；流式优先Kinesis + Lambda。
• 依赖复杂度：单步骤任务不需要Step Functions；多步骤、有分支、有重试需求时，Step Functions 的可见性和容错能力远超在代码里写if-else。
• 元数据管理：只要涉及多张表关联和Schema演化，Glue Data Catalog 几乎是必选项，EMR和Athena都可以直接读取。

一个可参考的端到端配置清单

以下清单可用于检查一条特征工程流水线是否完整。字段和参数根据实际项目调整：

最后需要强调：这四个服务不是互斥关系，而是分层协作。Step Functions 管“顺序和容错”，Glue 和 EMR 管“算得动”，Lambda 管“接得住事件”。在特征工程中，最常见的反模式是把所有逻辑塞进一个巨大的Glue Job脚本，导致失败后无法定位、重跑代价高。合理的做法是按职责拆分，用Step Functions显式表达依赖，让每个环节的输出都可独立验证。