More than 3 years have passed since last update.

Glueの使い方的な㊶(Workflows/ジョブ間でパラメータを受け渡す)

Last updated at 2020-05-11Posted at 2020-05-11

Glue Workflows とは

以下のリンクをご参照ください

Glueの使い方的な㊵(Workflowsでジョブフローの可視化)
https://qiita.com/pioho07/items/0cd0ae27b61f5914f78d

Workflowsを作る

内容

以下の記事で書いたクローラーやジョブを使いワークフローを作ります
https://qiita.com/pioho07/items/a24d188d67fe97034b34

クローラー -> ジョブ(PySpark) -> ジョブ(PythonShell)

処理はシンプルで、S3のcsvファイルをクローリングし、parquet変換し、変換後のファイル名をリネームします。これらの処理でワークフローから環境変数を表示し、渡し、その値に変更を加えます。後続の処理でその環境変数を取得し表示します。ジョブ間でのパラメータの受け渡すようなイメージです。

1つ目のクローラー：S3のcsvファイルをクローリングしGlue Data Catalogのテーブル(スキーマ)を作る
2つ目のジョブ：PySparkでフォーマットをparquetにしたり、country,year,month,day,hourでパーティション化したり、圧縮してS3に出力
3つ目のジョブ：PythonShellで出力されたファイルをリネームする

※詳細なコードの内容は上のリンクを参照ください

全体の流れ

前準備
ワークフロー作成
ワークフロー実行
確認

前準備

リソース名

クローラー名

se2_in0

ジョブ名

se2_job24(job15の微修正してパラメータを渡す)
se2_job25(job16の微修正してパラメータを渡す)

コード修正部分

se2_job24

コードの中で「sc = SparkContext()」より上の部分を修正しています。

getResolvedOptionsで'WORKFLOW_NAME'と'WORKFLOW_RUN_ID'も取得しています。get_workflow_run_propertiesでこの値を使って、ワークフローの実行プロパティを取得しています。また、直後にput_workflow_run_propertiesで新しいprodという値を入れてworkflow_envの値を更新しています。

se2_job24

import sys
#add start
import boto3
#add end

from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

###delete
#args = getResolvedOptions(sys.argv, ['JOB_NAME'])

###add start
glue_client = boto3.client("glue", region_name='ap-northeast-1')
args = getResolvedOptions(sys.argv, ['JOB_NAME', 'WORKFLOW_NAME', 'WORKFLOW_RUN_ID'])
workflow_name = args['WORKFLOW_NAME']
workflow_run_id = args['WORKFLOW_RUN_ID']
workflow_params = glue_client.get_workflow_run_properties(Name=workflow_name, RunId=workflow_run_id)["RunProperties"]
run_env = workflow_params['workflow_env']
print(run_env)
## put
workflow_params['workflow_env'] = 'prod'
glue_client.put_workflow_run_properties(Name=workflow_name, RunId=workflow_run_id, RunProperties=workflow_params)
###add end

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 = "se2", table_name = "se2_in0", transformation_ctx = "datasource0")
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("deviceid", "string", "deviceid", "string"), ("uuid", "long", "uuid", "long"), ("appid", "long", "appid", "long"), ("country", "string", "country", "string"), ("year", "long", "year", "long"), ("month", "long", "month", "long"), ("day", "long", "day", "long"), ("hour", "long", "hour", "long")], transformation_ctx = "applymapping1")
resolvechoice2 = ResolveChoice.apply(frame = applymapping1, choice = "make_struct", transformation_ctx = "resolvechoice2")
dropnullfields3 = DropNullFields.apply(frame = resolvechoice2, transformation_ctx = "dropnullfields3")

df = dropnullfields3.toDF()

partitionby=['country','year','month','day','hour']
output='s3://test-glue00/se2/out15/'
codec='snappy'

df.repartition(1).write.partitionBy(partitionby).mode("overwrite").parquet(output,compression=codec)
job.commit()

se2_job25

「s3 = boto3.resource('s3')」より上の部分を修正しています。１つ目のジョブと同じようにgetResolvedOptionsで'WORKFLOW_NAME'と'WORKFLOW_RUN_ID'を取得し、get_workflow_run_propertiesでこの値を使って、ワークフローの実行プロパティを取得しています。この取得した実行プロパティが１つ目のジョブで更新した値になっていることを後半で確認します。

se2_job25

# -*- coding: utf-8 -*-
import boto3
import re

###add start
import sys
from awsglue.utils import getResolvedOptions

glue_client = boto3.client("glue", region_name='ap-northeast-1')
args = getResolvedOptions(sys.argv, ['WORKFLOW_NAME', 'WORKFLOW_RUN_ID'])
workflow_name = args['WORKFLOW_NAME']
workflow_run_id = args['WORKFLOW_RUN_ID']
workflow_params = glue_client.get_workflow_run_properties(Name=workflow_name, RunId=workflow_run_id)["RunProperties"]
run_env = workflow_params['workflow_env']
print(run_env)
###add end

s3 = boto3.resource('s3')
bucket = s3.Bucket('test-glue00')
bucket_name='test-glue00'
for object in bucket.objects.filter(Prefix='se2/tmp2/country='):
   #print(object.key)
   old_file = object.key

   pattern1 = r'.*part.*'
   result1 = re.match(pattern1, old_file)
   if result1:
       Copy_from = result1.group()
       Copy_to = result1.group().rsplit('/', 1)[0] + '/' + result1.group().split("/")[2]
       s3.Object(bucket_name,Copy_to).copy_from(CopySource=bucket_name + '/' + Copy_from )
       s3.Object(bucket_name,Copy_from).delete()

ワークフロー作成

Glueの画面の左メニューから"ワークフロー"をクリックし[ワークフローの追加]をクリック

以下を入力して[ワークフローの追加]をクリックする
ここで入力しているプロパティは"実行プロパティ"と呼ばれ、ジョブの中で読み出して使い、また更新することができる値。ジョブ間で受け渡しできる環境変数のようなもの。今回はworkflow_env:devというキーバリューで(この値には特に意味はないです)、1つ目のジョブでこの値を読みworkflow_evn:prodに更新する。２つ目のジョブでworkflow_envを読みprodの値を得る。ということをやってみる。

ワークフロー名: se2_workflow2
[プロパティの追加]をクリックし以下を入力
キー：env
値：test