教學課程：撰寫 Spark 指令碼的 AWS Glue

本教學將向您介紹撰寫 AWS Glue 指令碼的程序。您可以按照任務的排程執行指令碼，也可以與互動式工作階段互動執行。如需任務的詳細資訊，請參閱建立視覺ETL工作 AWS Glue Studio。如需互動式工作階段的詳細資訊，請參閱概觀 AWS Glue 互動式工作階段。

AWS Glue Studio 視覺化編輯器提供無程式碼的圖形介面，可用於建置 AWS Glue 工作。 AWS Glue 腳本回視覺作業。您可透過這些指令碼存取一組經過擴增的工具，這些工具能與 Apache Spark 程式搭配運作。您可以存取原生 Spark API，以及 AWS Glue 程式庫，以便從 G AWS lue 指令碼中擷取、轉換和載入 (ETL) 工作流程。

本教學課程中，您會擷取、轉換及載入違停罰單資料集。執行這項工作的指令碼在形式和功能上與 AWS 大數據部落格上使用 AWS Glue Studio 讓 ETL 更容易產生的指令碼相同，後者介紹了 AWS Glue Studio 視覺化編輯器。透過在工作中執行此指令碼，您可以將其與視覺化工作進行比較，並查看 AWS Glue ETL 指令碼的運作方式。如此一來，您就可以做好準備，以便日後使用目前視覺化任務尚未提供的其他功能。

在本教學課程中，您將使用 Python 語言和程式庫。類似的功能在 Scala 中可用。在完成本教程之後，您應該能夠生成並檢查 Scala 示例腳本，以了解如何執行 Scala AWS Glue ETL 腳本編寫過程。

必要條件

本教學課程具備下列先決條件：

產生範例指令碼

您可以使用 AWS 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. 建立任務並貼上指令碼

在此步驟中，您要在中建立 AWS Glue 工作 AWS Management Console。這會設定允許 AWS Glue 執行指令碼的組態。此操作還會為您建立一個存放和編輯指令碼的地方。

步驟 2. 匯入 AWS Glue 程式庫

您需要設定指令碼，使其與在指令碼外部定義的程式碼和組態互動。這項工作是在 AWS Glue 工作室幕後完成的。

在此步驟中，您會執行下列動作。

在此程序中，您需編寫下列程式碼。此程式碼是產生的範例指令碼的一部分。

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 程序中，您都需要先定義要變更的來源資料集。在 AWS Glue Studio 視覺化編輯器中，您可以透過建立來源節點來提供此資訊。

在此步驟中，您需為 create_dynamic_frame.from_catalog 方法提供 database 和 table_name，以便從 AWS 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. 使用 AWS Glue 來轉換資料

在 ETL 程序中擷取來源資料後，您需要說明要如何變更資料。您可以在 AWS Glue Studio 視覺化編輯器中建立轉換節點來提供此資訊。

在此步驟中，您需為 ApplyMapping 方法提供目前所需欄位名稱和類型的映射，以轉換您的 DynamicFrame。

執行以下轉換。

create_dynamic_frame.from_catalog 將為您提供 DynamicFrame 物件。A DynamicFrame 代表 AWS Glue 中的資料集。 AWS Glue 轉換是改變的操作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. 將資料載入目標

轉換資料之後，通常需將轉換後的資料存放在與來源不同的地方。您可以在 AWS Glue Studio 視覺化編輯器中建立目標節點來執行此作業。

在此步驟中，您需為 write_dynamic_frame.from_options 方法提供 connection_type、connection_options、format 和 format_options，將資料載入 Amazon S3 中的目標儲存貯體。

在步驟 1 中，您初始化了 GlueContext 物件。在 AWS 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 物件。您需要在指令碼結尾手動斷定其生命週期，某些選用功能需要此操作才能正常執行。這項工作是在 AWS Glue 工作室幕後完成的。

在此步驟中，在 Job 物件上呼叫 commit 方法。

在此程序中，您需編寫下列程式碼。此程式碼是產生的範例指令碼的一部分。

job.commit()

選用 – 啟用任務書籤

之前的每個步驟都已經指示您設定 transformation_ctx 參數。這與名為任務書籤的功能有關。

藉由任務書籤，您可針對資料集定期執行任務，達到節省時間和金錢的目的，同時還能輕鬆追蹤先前的工作。Job 書籤會追蹤先前執行資料集中 AWS Glue 轉換的進度。通過跟踪先前運行結束的位置， AWS Glue 可以將其工作限制為以前未處理的行。如需任務書籤的詳細資訊，請參閱 使用任務書籤追蹤處理的資料。

若要啟用任務書籤，請先新增 transformation_ctx 陳述式至我們所提供的函數，如之前的範例所述。任務書籤狀態會在執行期間維持不變，而 transformation_ctx 參數是存取該狀態所需的索引鍵。這些陳述式自身不會執行任何操作。您還需要在任務的組態中啟用該功能。

在此程序中，您會使用 AWS Management Console啟用任務書籤。

步驟 7. 將程式碼作為任務執行

在此步驟中，您需執行任務以確認您是否已成功完成本教學課程。這是通過點擊一個按鈕來完成的，就像在 AWS Glue Studio 可視化編輯器中一樣。

如需有關設定和管理任務的詳細資訊，請參閱 提供您的自訂指令碼。

其他資訊

阿帕奇星火庫和方法在 AWS Glue 腳本中可用。您可以查看 Spark 文件以了解您可以使用這些包含的程式庫來做什麼。如需詳細資訊，請參閱 Spark 來源儲存庫的範例區段。

AWS Glue 2.0 + 默認情況下包括幾個常見的 Python 庫。在 Scala 或 Python 環境中，還有一些機制可以將自己的依賴項加載到 AWS Glue 作業中。如需 Python 相依性的相關資訊，請參閱 以 AWS Glue 使用 Python 程式庫。

有關如何在 Python 中使用 AWS Glue 功能的更多示例，請參閱AWS Glue Python 程式碼範例。Scala 和 Python 任務具有相同的功能，所以我們的 Python 範例應能為您提供以 Scala 執行類似工作的些許想法。