如何通过CloudFormation模板创建AWS Glue Job脚本并实现Apply Mapping功能

一、如何在CloudFormation模板中实现Apply Mapping功能 #

CloudFormation没有提供和控制台可视化Apply Mapping对应的直接配置项——因为控制台的Apply Mapping本质是帮你自动生成包含字段映射逻辑的PySpark/Scala代码。要在CFN模板中实现该功能，有两种可行方案：

1. 手动编写包含映射逻辑的脚本并引用 #

直接编写包含apply_mapping方法（PySpark）或对应Scala代码的Glue脚本，上传到S3后，在CFN模板的Glue Job资源中通过Command.ScriptLocation指定脚本路径。

示例CFN模板片段：

Resources:
  MyGlueJob:
    Type: AWS::Glue::Job
    Properties:
      Name: MyMappingJob
      Role: !Ref GlueJobRole
      Command:
        Name: glueetl
        ScriptLocation: s3://your-bucket/path/to/mapping-script.py
        PythonVersion: "3"
      DefaultArguments:
        "--job-language": python

对应的PySpark脚本示例（包含Apply Mapping逻辑）：

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

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

# 读取数据源
datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "your-db", table_name = "source-table", transformation_ctx = "datasource0")

# 执行Apply Mapping
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [
    ("source_col1", "string", "target_col1", "string"),
    ("source_col2", "int", "target_col2", "int")
], transformation_ctx = "applymapping1")

# 写入目标表
datasink2 = glueContext.write_dynamic_frame.from_catalog(frame = applymapping1, database = "your-db", table_name = "target-table", transformation_ctx = "datasink2")

job.commit()

2. 使用内联脚本（适合简单场景） #

如果脚本逻辑较短，可以直接在CFN模板中用Command.InlineScript嵌入代码，无需上传到S3：

Resources:
  MyInlineGlueJob:
    Type: AWS::Glue::Job
    Properties:
      Name: MyInlineMappingJob
      Role: !Ref GlueJobRole
      Command:
        Name: glueetl
        InlineScript: |
          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

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

          datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "your-db", table_name = "source-table", transformation_ctx = "datasource0")
          applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("source_col1", "string", "target_col1", "string")], transformation_ctx = "applymapping1")
          datasink2 = glueContext.write_dynamic_frame.from_catalog(frame = applymapping1, database = "your-db", table_name = "target-table", transformation_ctx = "datasink2")

          job.commit()
        PythonVersion: "3"
      DefaultArguments:
        "--job-language": python

二、CloudFormation创建Glue Job时是否会自动生成脚本？ #

不会自动生成。控制台的脚本自动生成依赖于你在界面上选择数据源、配置字段映射等交互操作，而CloudFormation是声明式基础设施即代码工具，没有交互流程触发脚本生成。要实现类似控制台的自动生成效果，可采用以下两种方式：

1. 导出控制台生成的脚本复用 #

先通过控制台创建一个包含所需Apply Mapping逻辑的Glue Job，导出控制台自动生成的脚本，上传到S3存储桶后，在CloudFormation模板中通过Command.ScriptLocation引用该脚本路径即可。

2. 自定义资源调用Glue GenerateScript API（进阶方案） #

如果你需要完全通过代码自动生成脚本，可以借助CloudFormation自定义资源，调用AWS Glue的GenerateScript API，传入数据源、映射规则等参数，生成脚本后将其存储到S3，再关联到Glue Job资源。这种方式需要编写Lambda函数作为自定义资源的后端处理逻辑，适合复杂或需要动态生成脚本的场景。

内容的提问来源于stack exchange，提问作者aws-beginner