Python での AWS Glue API の呼び出し

Boto 3 リソース API は AWS Glue にはまだ使用できないことに注意してください。現時点では、Boto 3 クライアント API のみ使用することができます。

Python の AWS Glue API 名

Java や他のプログラミング言語での AWS Glue API 名は、通常 CamelCased です。ただし、Python から呼び出されるとき、これらの一般名は、より「Python 的」にするために小文字に変更され、名前の一部がアンダースコア文字で区切られます。AWS Glue API リファレンスドキュメントでは、これらの Python 用の名前を一般的な CamelCased 形式の名前の後に括弧で囲んで一覧表示しています。

ただし、AWS Glue API 名自体は小文字に変換されますが、パラメータ名は大文字のままです。次のセクションで説明されますが、AWS Glue API の呼び出し時にパラメータを名前で渡す必要があるため、この点を覚えておくことが重要です。

AWS Glue の Python パラメータの受け渡しとアクセス

AWS Glue API の Python 呼び出しでは、明示的に名前でパラメータを渡すことが最善です。例:

job = glue.create_job(Name='sample', Role='Glue_DefaultRole',
                      Command={'Name': 'glueetl',
                               'ScriptLocation': 's3://my_script_bucket/scripts/my_etl_script.py'})

Job 構造 または JobRun の構造 の ETL スクリプトに渡す引数として指定する名前と値のタプルのディクショナリを Python が 作成することを理解しておくことは役立ちます。その後 Boto 3 が REST API 呼び出しを経由して JSON 形式でそれらを AWS Glue に渡します。つまり、スクリプトでそれらにアクセスするときは、引数の順序に依存することはできません。

たとえば、Python Lambda ハンドラ関数で JobRun を開始しようとしており、複数のパラメータを指定するとします。コードは以下のようになります。

from datetime import datetime, timedelta

client = boto3.client('glue')

def lambda_handler(event, context):
  last_hour_date_time = datetime.now() - timedelta(hours = 1)
  day_partition_value = last_hour_date_time.strftime("%Y-%m-%d")
  hour_partition_value = last_hour_date_time.strftime("%-H")

  response = client.start_job_run(
               JobName = 'my_test_Job',
               Arguments = {
                 '--day_partition_key':   'partition_0',
                 '--hour_partition_key':  'partition_1',
                 '--day_partition_value':  day_partition_value,
                 '--hour_partition_value': hour_partition_value } )

これらのパラメータに ETL スクリプトで確実にアクセスするには、AWS Glue の getResolvedOptions 関数を使用して名前で指定し、その後作成されたディクショナリからアクセスします。

import sys
from awsglue.utils import getResolvedOptions

args = getResolvedOptions(sys.argv,
                          ['JOB_NAME',
                           'day_partition_key',
                           'hour_partition_key',
                           'day_partition_value',
                           'hour_partition_value'])
print "The day partition key is: ", args['day_partition_key']
print "and the day partition value is: ", args['day_partition_value']

ネストされた JSON 文字列により AWS Glue ETL ジョブに引数を渡す際は、そのパラメータ値を保持するために、ジョブの実行が開始される前にパラメータ文字列をエンコードし、ジョブスクリプトが参照される前にパラメータ文字列をデコードする必要があります。例えば、次の引数文字列を考えてみます。

glue_client.start_job_run(JobName = "gluejobname", Arguments={
"--my_curly_braces_string": '{"a": {"b": {"c": [{"d": {"e": 42}}]}}}'
})

このパラメータを正しく渡すには、引数を Base64 の文字列としてエンコードする必要があります。

import base64
...
sample_string='{"a": {"b": {"c": [{"d": {"e": 42}}]}}}'
sample_string_bytes = sample_string.encode("ascii")

base64_bytes = base64.b64encode(sample_string_bytes) 
base64_string = base64_bytes.decode("ascii") 
...
glue_client.start_job_run(JobName = "gluejobname", Arguments={
"--my_curly_braces_string": base64_bytes})
...
sample_string_bytes = base64.b64decode(base64_bytes) 
sample_string = sample_string_bytes.decode("ascii") 
print(f"Decoded string: {sample_string}") 
...

例: ジョブを作成し実行する

以下の例では、Python を使用して AWS Glue API を呼び出し、ETL ジョブを作成して実行する方法を示します。

ジョブを作成し実行するには

1. AWS Glue クライアントのインスタンスを作成します。

   import boto3
   glue = boto3.client(service_name='glue', region_name='us-east-1',
                 endpoint_url='https://glue.us-east-1.amazonaws.com')

2. ジョブを作成します。次のコードに示すように、ETL コマンドの名前として glueetl を使用する必要があります。

   myJob = glue.create_job(Name='sample', Role='Glue_DefaultRole',
                             Command={'Name': 'glueetl',
                                      'ScriptLocation': 's3://my_script_bucket/scripts/my_etl_script.py'})

3. 前のステップで作成したジョブの新しい実行を開始します。

   myNewJobRun = glue.start_job_run(JobName=myJob['Name'])

4. ジョブのステータスを取得します。

   status = glue.get_job_run(JobName=myJob['Name'], RunId=myNewJobRun['JobRunId'])

5. ジョブ実行の現在の状態を出力します。

   print(status['JobRun']['JobRunState'])