GlueのWorkflows ジョブ間でパラメータを取得

getResolvedOptionsで'WORKFLOW_NAME'と'WORKFLOW_RUN_ID'も取得しています。get_workflow_run_propertiesでこの値を使って、ワークフローの実行プロパティを取得しています。

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

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

---

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

---

ジョブパラメータの取得方法
https://qiita.com/YujiHamada3/items/780af5c7bb5b9d4fa997
import sys
from awsglue.utils import getResolvedOptions
args = getResolvedOptions(sys.argv, ['hoge_string', 'hoge_int'])
step_count = args['hoge_string']
store_name = args['hoge_int']