Pythonのboto3でAWS Glueデータカタログのワークフローを開始する方法

本記事では、boto3ライブラリを使って、AWS Glueデータカタログ上でワークフローを開始（実行）する方法を解説します。サンプルコード付きなので、そのままコピーして動作を確認できます。

課題の定義

Pythonのboto3ライブラリを使用して、指定したワークフローを開始する関数を実装します。

解決のためのアプローチと手順

• ステップ1： boto3をインポートし、例外処理のためにbotocore.exceptionsからClientErrorも読み込みます。

• ステップ2： 関数の引数としてworkflow_name（ワークフロー名）を受け取ります。

• ステップ3： boto3のSessionクラスを使ってAWSセッションを作成します。デフォルトプロファイルにregion_nameが設定されていることを確認してください。設定されていない場合は、セッション作成時に明示的にregion_nameを渡す必要があります。

• ステップ4： glueサービス用のAWSクライアントを作成します。

• ステップ5： start_workflow_runメソッドを呼び出し、パラメータNameにworkflow_nameを指定します。

• ステップ6： レスポンスとしてRunId（実行ID）とレスポンスメタデータが返却され、ワークフローの実行が開始されます。

• ステップ7： ワークフロー開始時に問題が発生した場合に備え、汎用的な例外処理を実装します。

サンプルコード

以下のコードは、ワークフローを開始する例です。

import boto3
from botocore.exceptions import ClientError

def start_a_workflow(workflow_name):
    session = boto3.session.Session()
    glue_client = session.client('glue')
    try:
        response = glue_client.start_workflow_run(Name=workflow_name)
        return response
    except ClientError as e:
        raise Exception("boto3 client error in start_a_workflow: " + e.__str__())
    except Exception as e:
        raise Exception("Unexpected error in start_a_workflow: " + e.__str__())

print(start_a_workflow("test-daily"))

実行結果

コードを実行すると、以下のようにRunIdとレスポンスメタデータが出力されます。HTTPステータスコード200が返っていれば、ワークフローが正常に開始されたことを意味します。

{'RunId': 'wr_64e880240692fddd5e1b19aed587f856bc20a96f54bc', 'ResponseMetadata': {'RequestId': '782e953b-8ee3-4876-9b2c-cd35e147b513', 'HTTPStatusCode': 200, 'HTTPHeaders': {'date': 'Sun, 28 Mar 2021 08:11:02 GMT', 'content-type': 'application/x-amz-json-1.1', 'content-length': '79', 'connection': 'keep-alive', 'x-amzn-requestid': '782e953b-********************************13'}, 'RetryAttempts': 0}}

まとめ

boto3のstart_workflow_runメソッドを使えば、わずか数行のコードでAWS Glueのワークフローをプログラムから起動できます。返却されるRunIdを利用すれば、後続の処理でワークフローの実行状況をget_workflow_runメソッドなどで監視することも可能です。ETLジョブの自動化や定期実行の仕組みづくりにぜひ活用してください。