Glue Studio を Script 化し Job Parameter を与えて、Step Functions から Glue Job を実行してみた

# はじめに

Glue を使ったデータ加工を 1 日 1 回 行うときに、出力先の S3 に日付ごとにディレクトリを分けたいときがあります。この時に、Glue の Job Parameter を使うことで、出力先ディレクトリなどのパラメーターを渡す機能があり、こちらで柔軟に指定が可能です。

ただ、2023 年 8 月現在、Glue Studio の GUI を使ってデータ加工を行っている場合は、Job Parameter を使って S3 のディレクトリを指定することが出来ません。GUI を使ったデータ加工フローを Script 化すると、柔軟に指定が出来るようになります。今回はその手順を紹介していきます。

また、Glue を含めた全体のワークフローとして Step Functions を利用頂くことも多くあります。今回は、Step Functions から Glue を呼びだすときのパラメーターを指定する方法も紹介していきます。



# 前提

Glue Studio を使って、データ加工のフローを持っている前提で記事を記載します。なお、このフローは [前回の記事](https://qiita.com/sugimount-a/items/65ecd9fe14f2c3505a70) で作成しました。

![image-20230812221556376.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/80049350-9d43-b8f4-9d1b-500eb99996e0.png)



# Glue : Copy して Script 化

Glue Studio で作成したフローを Script 化していきます。**注意点があります。GUI ベースのものを Script 化にしてしまうと、GUI に戻すことはできません。そのため、Script 化する前に、そのフローを Clone してバックアップを持っておきましょう。**

Script 化するフローを選択して、Clone job を選びます。

![image-20230812221633553.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/3decccec-cde6-caae-f702-857cfd18078c.png)



Clone した Job を使って、Edit Script を押します。

![image-20230812222025390.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/772f55b0-ab64-8ccc-a111-8a7a8bd02b09.png)



Confirm を押します。

![image-20230812222050431.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/6a97d261-3ab5-eee8-26f4-879df73ea217.png)



すると、Script が生成されました。全体は以下のようなコードになっています。

```python
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 awsglue.dynamicframe import DynamicFrame

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 transformcode
transformcode_node1691764083220 = glueContext.create_dynamic_frame.from_options(
    format_options={
        "quoteChar": '"',
        "withHeader": True,
        "separator": ",",
        "optimizePerformance": False,
    },
    connection_type="s3",
    format="csv",
    connection_options={
        "paths": ["s3://glue-studio-etltest01/codetransform.csv"],
        "recurse": True,
    },
    transformation_ctx="transformcode_node1691764083220",
)

# Script generated for node buhin
buhin_node1 = glueContext.create_dynamic_frame.from_options(
    format_options={
        "quoteChar": '"',
        "withHeader": True,
        "separator": ",",
        "multiline": False,
        "optimizePerformance": False,
    },
    connection_type="s3",
    format="csv",
    connection_options={
        "paths": ["s3://glue-studio-etltest01/buhin.csv"],
        "recurse": True,
    },
    transformation_ctx="buhin_node1",
)

# Script generated for node Join
buhin_node1DF = buhin_node1.toDF()
transformcode_node1691764083220DF = transformcode_node1691764083220.toDF()
Join_node1691764237797 = DynamicFrame.fromDF(
    buhin_node1DF.join(
        transformcode_node1691764083220DF,
        (buhin_node1DF["buhin"] == transformcode_node1691764083220DF["before"]),
        "outer",
    ),
    glueContext,
    "Join_node1691764237797",
)

# Script generated for node Drop Fields
DropFields_node1691764648382 = DropFields.apply(
    frame=Join_node1691764237797,
    paths=["before"],
    transformation_ctx="DropFields_node1691764648382",
)

# Script generated for node Rename Field
RenameField_node1691764779832 = RenameField.apply(
    frame=DropFields_node1691764648382,
    old_name="after",
    new_name="bufin_converted",
    transformation_ctx="RenameField_node1691764779832",
)

# Script generated for node Amazon S3
AmazonS3_node1691846031658 = glueContext.write_dynamic_frame.from_options(
    frame=RenameField_node1691764779832,
    connection_type="s3",
    format="csv",
    connection_options={
        "path": "s3://glue-studio-etltest01/output/",
        "partitionKeys": [],
    },
    transformation_ctx="AmazonS3_node1691846031658",
)

job.commit()
```



# Glue : Code を修正して、Job Parameter を受け取る

まず、修正後のコードを全て記載します。このあとに、差分を説明します。

```python
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 awsglue.dynamicframe import DynamicFrame

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

print("Debug!!!!!!!")
output_dir=args["output_dir"]
print(output_dir)
test_param=args["test_param"]
print(args["test_param"])

# Script generated for node transformcode
transformcode_node1691764083220 = glueContext.create_dynamic_frame.from_options(
    format_options={
        "quoteChar": '"',
        "withHeader": True,
        "separator": ",",
        "optimizePerformance": False,
    },
    connection_type="s3",
    format="csv",
    connection_options={
        "paths": ["s3://glue-studio-etltest01/codetransform.csv"],
        "recurse": True,
    },
    transformation_ctx="transformcode_node1691764083220",
)

# Script generated for node buhin
buhin_node1 = glueContext.create_dynamic_frame.from_options(
    format_options={
        "quoteChar": '"',
        "withHeader": True,
        "separator": ",",
        "multiline": False,
        "optimizePerformance": False,
    },
    connection_type="s3",
    format="csv",
    connection_options={
        "paths": ["s3://glue-studio-etltest01/buhin.csv"],
        "recurse": True,
    },
    transformation_ctx="buhin_node1",
)

# Script generated for node Join
buhin_node1DF = buhin_node1.toDF()
transformcode_node1691764083220DF = transformcode_node1691764083220.toDF()
Join_node1691764237797 = DynamicFrame.fromDF(
    buhin_node1DF.join(
        transformcode_node1691764083220DF,
        (buhin_node1DF["buhin"] == transformcode_node1691764083220DF["before"]),
        "outer",
    ),
    glueContext,
    "Join_node1691764237797",
)

# Script generated for node Drop Fields
DropFields_node1691764648382 = DropFields.apply(
    frame=Join_node1691764237797,
    paths=["before"],
    transformation_ctx="DropFields_node1691764648382",
)

# Script generated for node Rename Field
RenameField_node1691764779832 = RenameField.apply(
    frame=DropFields_node1691764648382,
    old_name="after",
    new_name="bufin_converted",
    transformation_ctx="RenameField_node1691764779832",
)

# Script generated for node Amazon S3
AmazonS3_node1691846031658 = glueContext.write_dynamic_frame.from_options(
    frame=RenameField_node1691764779832,
    connection_type="s3",
    format="csv",
    connection_options={
        "path": "s3://glue-studio-etltest01/output/" + output_dir + "/",
        "partitionKeys": [],
    },
    transformation_ctx="AmazonS3_node1691846031658",
)

job.commit()
```



引数を受けとる部分です。`getResolvedOptions` の関数で、output_dir と test_param の引数を新たに受け取って args に格納しています。その後、print 分などでデバッグ的に標準出力に出しています。

```python
args = getResolvedOptions(sys.argv, ["JOB_NAME","output_dir","test_param"])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args["JOB_NAME"], args)

print("Debug!!!!!!!")
output_dir=args["output_dir"]
print(output_dir)
test_param=args["test_param"]
print(args["test_param"])
```



その後、S3 バケットに出力する部分です。`path` に `output_dir` を指定することで、出力するパラメータ―を指定できるようにしています。

```python
# Script generated for node Amazon S3
AmazonS3_node1691846031658 = glueContext.write_dynamic_frame.from_options(
    frame=RenameField_node1691764779832,
    connection_type="s3",
    format="csv",
    connection_options={
        "path": "s3://glue-studio-etltest01/output/" + output_dir + "/",
        "partitionKeys": [],
    },
    transformation_ctx="AmazonS3_node1691846031658",
)
```



Save を押して保存します。

![image-20230812234203121.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/4736ed60-b741-3e73-3c86-fac1d45118e9.png)



# Glue : 単体で動作確認

AWS CLI から Glue Job を実行する際に、パラメーターを渡して動作確認をしてみます。`--arguments=` の指定で、2 つのパラメーターを渡しています。`output_dir` が S3 バケットで指定されるパラメーターになります。

```shell
aws glue start-job-run \
--job-name "Transform Code Scripting" \
--arguments='--output_dir="20230812", --test_param="this is test"'
```



実行した結果、Console 上でジョブの状態が確認できます。この詳細画面に、標準出力を確認するためのリンクがあります。Cloudwatch Logs で `Output_logs` を選択します。

![image-20230812225557333.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/1e611ad8-37d1-b748-6260-82c756779d4d.png)



Python コードに記載したデバッグ文字列が確認できます。パラメーターで指定した `20230812` や `this is test` が確認できました。

![image-20230812225641126.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/ed406401-034c-41f5-5ce3-cfcc6f8d56fa.png)



S3 バケット上で、パラメータで指定した名前でディレクトリが自動作成され、データ加工の結果が格納されました。

![image-20230813003744111.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/8b146cc7-3f81-691e-f6c9-f0019784bc53.png)



# Step Functions : Glue を呼びだすステートマシンを作成

Step Functions から Glue Job を呼びだしてパラメーターの指定をしてみます。

![image-20230812234631852.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/29c5a0fb-cf75-0529-5eb3-dabe349f0781.png)



Design your workflow visually で Next を押します。

![image-20230812235302634.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/31e854f9-3b65-5fb1-859c-ff1bbcb6ec62.png)



画面左側にある StartJobRun を検索して、真ん中に配置します。

![image-20230812235621910.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/12855340-73dd-15c4-ee29-6050425d02bb.png)



この Start Job Run の API Parameter を確認するために AWS Document を見ながら作り上げてもよいのですが、AWS CLI のデバッグからパラメーターを確認する方法もあります。AWS CLI の実行に `--debug` を付けて実行します。

```
aws glue start-job-run \
      --job-name "Transform Code Scripting" \
      --arguments='--output_dir="20230812", --test_param="this is test"' --debug
```



デバッグメッセージが表示されるなかで、以下の文に注目します。右にスクロールすると `body` のパラメータが見えます。これを Step Functions で同様に指定すれば OK です。

```
2023-08-12 23:58:57,959 - MainThread - botocore.endpoint - DEBUG - Making request for OperationModel(name=StartJobRun) with params: ...省略... 'body': b'{"JobName": "Transform Code Scripting", "Arguments": {"--output_dir": "20230812", "--test_param": "this is test"}}' 省略
```



Step Functions の API Parameter に次を指定します。

```json
{
  "JobName": "Transform Code Scripting",
  "Arguments": {
    "--output_dir": "20230813",
    "--test_param": "this is test"
  }
}
```

![image-20230813000645770.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/3a397a09-6028-e08e-a9e1-5361b7bbd5e3.png)



Glue の実行完了を待つために、Wait for task to complete にチェックを入れます。

![image-20230813002118002.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/f71ee120-ec26-ca8b-bf5e-4fd1731a5c81.png)



通常時は Success、エラー時は Fail につなげる設定を入れた後に、Next を押します。

![image-20230813001010786.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/3fe896ea-9f4e-0c8a-929a-a2de146454ff.png)



Next

![image-20230813001140094.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/a1eb1eff-9f12-b5a1-dd92-95aa6fb3c904.png)



適当に State Machine の名前などを入れて、Create を押します。

![image-20230813001248664.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/47294154-5ddf-dabb-268a-65989adb2bac.png)



# StepFunction : Glue を呼びだす実行確認

作成された State machine 上で Start execution を押します。

![image-20230813001518492.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/defba8ab-b68e-51d2-d8d3-fb2749106003.png)



Start execution を押します。

![image-20230813001533985.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/8ed3f76d-9ff0-92e8-852b-841498dfb899.png)



実行の様子

- Glue の Job 実行が終わるまで、待機している様子が見えます。

![image-20230813001656281.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/a2ca51b8-82e7-f182-78a1-1e5054efb16a.png)



実行が成功しました。

![image-20230813001818792.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/133c2502-8a61-afbc-e133-2a1638d06921.png)



S3 Bucket を確認すると、渡したパラメータが利用され、S3 の出力先ディレクトリ名として利用されています。

![image-20230813001854114.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1002774/051487f2-9f49-7638-ec99-7309a3132451.png)



# わかったこと

- Step Functions が Glue Job を実行するときに、実行が終わるまで待機することが簡単にできる
- Step Functions から、Glue Job へ Job Parameter を渡すことが可能



# 参考 URL

- Step Function を実行したときの日本日付を取得したい場合は、次の記事が参考になる
  - [Step Functions] ステートマシンで「日付」を取り扱う方法
  - https://dev.classmethod.jp/articles/step-functions-how-to-use-date-in-state-machine/