3. 批处理管道入门

在批处理数据管道中，AWS Glue 承担的是“把散落在存储里的原始数据，变成可查询、可分析的表”的核心角色。它由三个关键部件协同完成：Crawler 负责发现数据结构，Data Catalog 负责保存元数据，Job 负责执行真正的转换与写出。理解这三者的分工与数据流向，是构建第一个 ETL 作业的前提。

三个部件各自解决什么问题

Crawler 是一个自动化的元数据发现程序。你给它一个数据存储位置（例如 S3 前缀），它会按调度或手动触发去扫描文件，推断格式、分隔符、列名与数据类型，并把结果写入 Data Catalog。Data Catalog 是 Glue 的中央元数据仓库，它保存数据库、表、分区、列类型、SerDe 等信息，Athena、Redshift Spectrum、EMR 都依赖它来定位数据。Job 则是运行在 Glue 托管 Spark 环境上的 ETL 脚本，它从 Catalog 表或直接指定的路径读取数据，做清洗、转换、聚合，再写回 S3、JDBC 数据库或数据仓库。

三者的关系可以概括为：Crawler 产出元数据，Catalog 存储元数据，Job 消费元数据并产出新数据。下图展示了这一数据流与元数据流的分工。

动手前的准备：IAM 权限与 S3 布局

Glue 需要代表你访问 S3 和 Catalog，因此必须有一个具备相应权限的 IAM 角色。最小可用权限包括：对原始与目标 S3 前缀的 s3:GetObject、s3:PutObject、s3:ListBucket，以及对 Catalog 的 glue:GetDatabase、glue:GetTable、glue:CreateTable、glue:UpdateTable。AWS 提供了托管策略 AWSGlueServiceRole，再附加一个自定义的 S3 访问策略即可。

示例数据布局如下，假设我们有一份电商订单 CSV：

s3://demo-glue-pipeline/raw/orders/
  orders_2024_01.csv
  orders_2024_02.csv

列：order_id, customer_id, order_date, amount, status
示例行：1001,C023,2024-01-05,199.00,PAID

第一步：创建 Crawler 并运行

在 Glue 控制台选择 “Crawlers → Create crawler”，命名为 orders-raw-crawler。数据源选择 S3，路径填 s3://demo-glue-pipeline/raw/orders/。IAM 角色选择上一步创建的角色。目标数据库新建 sales_db，表名前缀可留空。调度先选 “On demand”，便于首次手动验证。

运行 Crawler 后，它会读取 CSV 的表头，推断出 order_id 为 bigint、amount 为 double、order_date 为 string、status 为 string。完成后在 Catalog 中会出现表 sales_db.orders。可以在 Athena 中执行 SELECT * FROM sales_db.orders LIMIT 10; 验证元数据是否正确。若列名出现 col0 这类默认名，通常说明文件没有表头或分隔符识别错误，需要在 Crawler 的 classifier 中调整。

第二步：编写并运行 Glue Job

创建 Job 时选择 “Spark script editor”，语言选 Python。Glue 会自动生成一段骨架代码，核心是 GlueContext 与 DynamicFrame。下面是一个可直接参考的脚本，它读取 Catalog 表、过滤无效订单、把日期转为 date 类型，并以 Parquet 分区写出：

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql.functions import col, to_date

args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# 从 Data Catalog 读取
dyf = glueContext.create_dynamic_frame.from_catalog(
    database="sales_db",
    table_name="orders",
    transformation_ctx="read_orders"
)

# 转为 Spark DataFrame 做清洗
df = dyf.toDF()
df_clean = (df
    .filter(col("amount") > 0)
    .filter(col("status").isin("PAID", "SHIPPED"))
    .withColumn("order_date", to_date(col("order_date"), "yyyy-MM-dd"))
)

# 写回 S3，按日期分区，Parquet 格式
out = glueContext.write_dynamic_frame.from_options(
    frame=DynamicFrame.fromDF(df_clean, glueContext, "clean"),
    connection_type="s3",
    connection_options={
        "path": "s3://demo-glue-pipeline/curated/orders/",
        "partitionKeys": ["order_date"]
    },
    format="parquet",
    transformation_ctx="write_orders"
)

job.commit()

Job 参数中需要设置 --JOB_NAME，并建议开启 “Job bookmark” 为 enable，这样重复运行时 Glue 会记录已处理的数据，避免重复写入。Worker 类型首次可选 G.1X、数量 2，足够处理示例数据。

第三步：监控运行与判断结果

Job 运行后，在 Glue 控制台的 “Runs” 标签可以看到状态从 STARTING → RUNNING → SUCCEEDED。关键观察点有三个：一是 运行时长与 DPU 使用，用于评估成本；二是 CloudWatch 日志，脚本中的 print 与异常堆栈都会输出到 /aws-glue/jobs/output 和 /aws-glue/jobs/error；三是 输出路径，成功后在 s3://demo-glue-pipeline/curated/orders/ 下应出现 order_date=2024-01-05/ 这类分区目录，内部是 Parquet 文件。

从第一个作业到可复用的管道

完成上述三步后，你已经拥有了一条最小可用的批处理链路：Crawler 发现原始表 → Catalog 保存元数据 → Job 清洗并写出 Parquet → 再次 Crawler 让结果可查询。后续要把它变成生产级管道，通常还需要：用 Glue Trigger 或 Step Functions 把 Crawler 与 Job 串成按天调度；把 Job 脚本放入 S3 并做版本管理；对输出表做分区维护与压缩；以及在 CloudWatch 上为失败次数和运行时长设置告警。第一个作业的价值不在于它多复杂，而在于它把“元数据驱动 ETL”这条主线跑通了。