チュートリアル: AWS Glue for Spark スクリプトの記述

このチュートリアルでは、AWS Glue スクリプトを記述するプロセスの概要を説明します。スクリプトを、ジョブを使用してスケジュールに従って実行したり、インタラクティブセッションを使用してインタラクティブに実行したりできます。ジョブの詳細については、「AWS Glue Studio でビジュアル ETL ジョブを作成する」を参照してください。詳細については、「AWS Glue インタラクティブセッションの概要」を参照してください。

AWS Glue Studio ビジュアルエディターは、AWS Glue ジョブを構築するためのグラフィカルなノーコードのインターフェイスを提供します。AWSGlue スクリプトはビジュアルジョブをバックアップします。Glue スクリプトを使用すると、Apache Spark プログラムを操作するための拡張された一連のツールにアクセスできます。AWS Glue スクリプト内からの抽出、変換、ロード (ETL) ワークフローを円滑にする AWS Glue ライブラリだけでなく、Spark の ネイティブ API にもアクセスできます。

このチュートリアルでは、駐車チケットのデータセットを抽出、変換、ロードします。この作業を行うスクリプトは、AWS Glue Studio ビジュアルエディタを紹介する AWS ビッグデータブログの「AWS Glue Studio で ETL を簡単にする」で作成したものと形式および機能が同じです。このスクリプトをジョブ内で実行することで、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 Management Consoleで AWS Glue ジョブを作成します。これにより、AWS Glue でスクリプトを実行するのに必要な構成がセットアップされます。また、スクリプトを保存および編集するための場所も作成されます。

ステップ 2. AWS Glue ライブラリをインポートする

スクリプトの外部で定義されたコードや設定とやり取りするには、スクリプトの設定が必要です。この作業は、AWS 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 プロセスにおいて、変更するソースデータセットを最初に定義する必要があります。AWS Glue Studio ビジュアルエディタでは、[Source] (ソース) ノードを作成することで、この情報を指定します。

このステップでは、AWS Glue データカタログで設定されたソースからデータを抽出する create_dynamic_frame.from_catalog メソッドに database および table_name を提供します。

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 ビジュアルエディタで [Transform] (変換) ノードを作成します。

このステップでは、希望する現在のフィールドの名前とタイプのマップを ApplyMapping メソッドに提供し、DynamicFrame を変換します。

以下の変換を実行します。

create_dynamic_frame.from_catalog により、DynamicFrame オブジェクトが提供されます。DynamicFrame は、AWS Glue 内のデータセットを表します。AWSGlue での変換は、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",
)

Step 5. ターゲットにデータをロードする

通常、変換後のデータは、そのソースとは別の場所に保存します。この操作を実行するには、AWS Glue Studio ビジュアルエディタで、[Target] (ターゲット) ノードを作成します

このステップでは、データを Amazon S3 のターゲットバケットにロードするための write_dynamic_frame.from_options メソッド connection_type、connection_options、format、format_options を提供します。

ステップ 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 オブジェクトのコミット

Job オブジェクトは、ステップ 1 で初期化してあります。スクリプトの終了時に、ライフサイクルを手動で閉じる必要があります。一部のオプション機能が正しく機能するために、これが必要となります。この作業は、AWS Glue Studioの背後で実行されます。

このステップでは、Job オブジェクトの commit メソッドを呼び出します。

この手順では、次のコードを作成します。このコードは、生成されるサンプルスクリプトの一部です。

job.commit()

オプション — ジョブのブックマークを有効にする

これまでのすべてのステップでは、transformation_ctx パラメータを設定することを促しています。これは、ジョブのブックマークと呼ばれる機能に関連した設定です。

ジョブのブックマークを使用すると、以前の作業を簡単に追跡できるデータセットに対して定期的に実行されるジョブによって、時間とコストを節約できます。ジョブのブックマークは、前回の実行のデータセット全体に関する AWS Glue 変換の進捗状況を追跡します。以前の実行が終了した場所を追跡することで、AWS Glue は未処理の行に対してのみ作業できます。ブックマークの詳細については、「ジョブのブックマークを使用した処理済みデータの追跡」を参照してください。

前の例で説明したように、ジョブのブックマークを有効にするには、始めに、提供されている関数に transformation_ctx ステートメントを追加します。ジョブのブックマークの状態は実行後も保持されます。transformation_ctx パラメータは、その状態にアクセスするために使用されるキーです。これらのステートメント単体では機能を持ちません。また、ジョブのための設定内で、その機能を有効にする必要があります。

この手順では、AWS Management Console を使用してジョブのブックマークを有効にします。

ステップ 7。コードをジョブとして実行する

このステップでは、チュートリアルを正常に完了したことを確認するために、ジョブを実行します。これは、AWS Glue Studio ビジュアルエディタの中で、ボタンをクリックするだけで完了します。

ジョブの設定と管理の詳細については、「独自のカスタムスクリプトの提供」を参照してください。

詳細情報

Apache Spark のライブラリとメソッドを、AWS Glue スクリプトの中で使用することが可能です。Spark のドキュメントを参照することで、これらの付属ライブラリで何が行えるのか把握できます。詳細については、「examples section of the Spark source repository」(Spark ソースリポジトリのサンプルセクション) を参照してください。

AWS Glue 2.0+ には、デフォルトで、いくつかの一般的な Python ライブラリが含まれています。Scala または Python 環境の AWS Glue ジョブに独自の依存関係をロードするためのメカニズムもあります。Python の依存関係については、「AWS Glue での Python ライブラリの使用」を参照してください。

Python で AWS Glue の機能を使用する方法の例の詳細については、「AWS Glue Python コードサンプル」を参照してください。Scala と Python ジョブには同等の機能があるため、Python の例を基にして、同様の処理を Scala で行う方法についても考察することが可能です。