教程：编写 Amazon Glue for Spark 脚本

本教程向您介绍编写 Amazon Glue 脚本的过程。脚本可以与作业一起按计划运行，也可以与交互式会话交互运行。有关作业的更多信息，请参阅使用 Amazon Glue Studio 构建可视化 ETL 作业。有关交互式会话的更多信息，请参阅 Amazon Glue 交互式会话概览。

Amazon Glue Studio 可视化编辑器提供图形化的零代码接口，用于构建 Amazon Glue 作业。AmazonGlue 脚本支持可视化作业。它们可让您访问用于使用 Apache Spark 程序的扩展工具集。您可以访问本机 Spark API，以及从 Amazon Glue 脚本访问为提取、转换、加载 (ETL) 工作流提供便利的 Amazon Glue 库。

在本教程中，您将提取、转换和加载停车罚单数据集。完成此工作的脚本在形式和功能上与 Amazon 大数据博客上的使用 Amazon Glue Studio 让 ETL 更轻松中生成的脚本完全相同，该博客介绍了 Amazon Glue Studio 可视化编辑器。通过在作业中运行此脚本，您可以将其与可视化作业进行比较，并了解 Amazon Glue ETL 脚本的工作原理。这可以让您为使用可视化作业中尚不可用的其他功能做好准备。

在本教程中，您使用 Python 语言和库。Scala 中提供了类似的功能。通读本教程后，您应该能够生成和检查示例 Scala 脚本，以了解如何进行 Scala Amazon Glue ETL 脚本编写过程。

先决条件

本教程包含以下先决条件：

生成示例脚本

您可以使用 Amazon Glue Studio 可视化编辑器作为强大的代码生成工具，为您要编写的脚本创建脚手架。您将使用此工具创建示例脚本。

如果您想跳过这些步骤，教程提供了此脚本。

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)

# Script generated for node S3 bucket
S3bucket_node1 = glueContext.create_dynamic_frame.from_catalog(
    database="yyz-tickets", table_name="tickets", transformation_ctx="S3bucket_node1"
)

# Script generated for node ApplyMapping
ApplyMapping_node2 = ApplyMapping.apply(
    frame=S3bucket_node1,
    mappings=[
        ("tag_number_masked", "string", "tag_number_masked", "string"),
        ("date_of_infraction", "string", "date_of_infraction", "string"),
        ("ticket_date", "string", "ticket_date", "string"),
        ("ticket_number", "decimal", "ticket_number", "float"),
        ("officer", "decimal", "officer_name", "decimal"),
        ("infraction_code", "decimal", "infraction_code", "decimal"),
        ("infraction_description", "string", "infraction_description", "string"),
        ("set_fine_amount", "decimal", "set_fine_amount", "float"),
        ("time_of_infraction", "decimal", "time_of_infraction", "decimal"),
    ],
    transformation_ctx="ApplyMapping_node2",
)

# Script generated for node S3 bucket
S3bucket_node3 = glueContext.write_dynamic_frame.from_options(
    frame=ApplyMapping_node2,
    connection_type="s3",
    format="glueparquet",
    connection_options={"path": "s3://DOC-EXAMPLE-BUCKET", "partitionKeys": []},
    format_options={"compression": "gzip"},
    transformation_ctx="S3bucket_node3",
)

job.commit()

第 1 步。创建作业并粘贴您的脚本

在此步骤中，您将在 Amazon Web Services Management Console 中创建一个 Amazon Glue 作业。这将设置允许 Amazon Glue 运行脚本的配置。它还为您创建了一个存储和编辑脚本的位置。

第 2 步。导入 Amazon Glue 库

您需要将脚本设置为与脚本外部定义的代码和配置进行交互。此工作在 Amazon Glue Studio 的后台完成。

在此步骤中，您执行以下操作。

在此过程中，您将编写以下代码。此代码是生成的示例脚本的一部分。

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)

第 3 步。从源提取数据

在任何 ETL 过程中，您首先需要定义要更改的源数据集。在 Amazon Glue Studio 可视化编辑器中，您可通过创建 Source（源）节点提供此信息。

在此步骤中，您为 create_dynamic_frame.from_catalog 方法提供 database 和 table_name，以从 Amazon Glue 数据目录中配置的源提取数据。

在上一步中，您初始化了 GlueContext 对象。可使用此对象查找用于配置源的方法，例如 create_dynamic_frame.from_catalog。

在此过程中，您将使用 create_dynamic_frame.from_catalog 编写以下代码。此代码是生成的示例脚本的一部分。

S3bucket_node1 = glueContext.create_dynamic_frame.from_catalog(
    database="yyz-tickets", table_name="tickets", transformation_ctx="S3bucket_node1"
    )

第 4 步。使用 Amazon Glue 转换数据

在 ETL 过程中提取源数据后，您需要描述想要如何更改数据。您可通过在 Amazon Glue Studio 可视化编辑器中创建 Transform（转换）节点来提供此信息。

在此步骤中，为 ApplyMapping 方法提供当前和所需字段名称和类型的映射以转换 DynamicFrame。

您可执行以下转换。

create_dynamic_frame.from_catalog 为您提供一个 DynamicFrame 对象。DynamicFrame 表示 Amazon Glue 中的数据集。AmazonGlue 转换是更改 DynamicFrames 的操作。

在此过程中，您将使用 ApplyMapping 编写以下代码。此代码是生成的示例脚本的一部分。

ApplyMapping_node2 = ApplyMapping.apply(
    frame=S3bucket_node1,
    mappings=[
        ("tag_number_masked", "string", "tag_number_masked", "string"),
        ("date_of_infraction", "string", "date_of_infraction", "string"),
        ("ticket_date", "string", "ticket_date", "string"),
        ("ticket_number", "decimal", "ticket_number", "float"),
        ("officer", "decimal", "officer_name", "decimal"),
        ("infraction_code", "decimal", "infraction_code", "decimal"),
        ("infraction_description", "string", "infraction_description", "string"),
        ("set_fine_amount", "decimal", "set_fine_amount", "float"),
        ("time_of_infraction", "decimal", "time_of_infraction", "decimal"),
    ],
    transformation_ctx="ApplyMapping_node2",
)

第 5 步。将数据加载到目标中

转换数据后，通常将转换后的数据存储在与源不同的位置。您可以通过在 Amazon Glue Studio 可视化编辑器中创建 target（目标）节点来执行此操作。

在此步骤中，您将为 write_dynamic_frame.from_options 方法提供 connection_type、connection_options、format 和 format_options 以将数据加载到 Amazon S3 中的目标桶中。

在步骤 1 中，您初始化了 GlueContext 对象。在 Amazon Glue 中，您将在此找到用于配置目标的方法，就像源一样。

在此过程中，您将使用 write_dynamic_frame.from_options 编写以下代码。此代码是生成的示例脚本的一部分。

S3bucket_node3 = glueContext.write_dynamic_frame.from_options(
    frame=ApplyMapping_node2,
    connection_type="s3",
    format="glueparquet",
    connection_options={"path": "s3://DOC-EXAMPLE-BUCKET", "partitionKeys": []},
    format_options={"compression": "gzip"},
    transformation_ctx="S3bucket_node3",
    )

第 6 步。提交 Job 对象

您在步骤 1 中初始化了 Job 对象。您需要在脚本结束时手动结束其生命周期。某些可选功能需要此操作才能正常运行。此工作在 Amazon Glue Studio 的后台完成。

在此步骤中，调用 commit 对象上的 Job 方法。

在此过程中，您将编写以下代码。此代码是生成的示例脚本的一部分。

job.commit()

可选 - 启用作业书签

在之前的每个步骤中，都会指示您设置 transformation_ctx 参数。这与称为作业书签的功能有关。

借助作业书签，您可以使用重复运行的作业节省时间和金钱，而不是使用可以轻松跟踪以前工作的数据集。作业书签跟踪 Amazon Glue 转换在之前运行的数据集上的进度。通过跟踪之前运行的结束位置，Amazon Glue 可以将其工作限制为之前未处理的行。有关作业书签的更多信息，请参阅 使用作业书签跟踪已处理的数据。

要启用作业书签，请先将 transformation_ctx 添加到提供的函数中，如前面的示例所述。作业书签状态在运行期间保持不变。transformation_ctx 参数是用于访问该状态的键。这些语句本身没有任何作用。您还需要激活作业配置中的功能。

在此过程中，您可以使用 Amazon Web Services Management Console 启用作业书签。

第 7 步。将代码作为作业运行

在此步骤中，运行作业以验证是否已成功完成本教程。这可以通过单击按钮完成，就像在 Amazon Glue Studio 可视化编辑器中一样。

有关配置和管理作业的更多信息，请查阅 提供您自己的自定义脚本。

更多信息

Apache Spark 库和方法在 Amazon Glue 脚本中提供。您可以查看 Spark 文档以了解如何使用这些随附的库。有关更多信息，请参阅 Spark 源存储库的示例部分。

Amazon Glue 2.0+ 默认包含多个常用的 Python 库。还有一些机制可以将您自己的依赖项加载到 Scala 或 Python 环境的 Amazon Glue 作业中。有关 Python 依赖项的信息，请参阅 将 Python 库与 Amazon Glue 一起使用。

有关如何在 Python 中使用 Amazon Glue 功能的更多示例，请参阅 Amazon Glue Python 代码示例。Scala 和 Python 作业具有同等的功能，因此我们的 Python 示例应该会让您了解如何在 Scala 中进行类似的工作。