複数のジョブの進行状況のモニタリング

複数の AWS Glue ジョブをまとめてプロファイルし、それらの間のデータフローを監視できます。これは、一般的なワークフローパターンであり、個々のジョブの進行状況、データ処理バックログ、データの再処理、ジョブのブックマークのモニタリングが必要です。

プロファイルされたコード

このワークフローには、入力ジョブと出力ジョブの 2 つがあります。入力ジョブは、定期的なトリガーを使用して 30 分ごとに実行するようにスケジュールされます。出力ジョブは、入力ジョブの実行が成功するたびに実行されるようにスケジュールされます。このようなスケジュールされたジョブは、ジョブトリガーを使用して制御されます。

入力ジョブ: このジョブは Amazon Simple Storage Service (Amazon S3) の場所からデータを読み取り、ApplyMapping を使用して変換し、Amazon S3 のステージング場所に書き込みます。次のコードは、入力ジョブのプロファイルされたコードです。

datasource0 = glueContext.create_dynamic_frame.from_options(connection_type="s3", connection_options = {"paths": ["s3://input_path"], "useS3ListImplementation":True,"recurse":True}, format="json")
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [map_spec])
datasink2 = glueContext.write_dynamic_frame.from_options(frame = applymapping1, connection_type = "s3", connection_options = {"path": staging_path, "compression": "gzip"}, format = "json")

出力ジョブ: このジョブは、Amazon S3 内のステージング場所から入力ジョブの出力を読み取り、もう一度変換して、ターゲットの場所に書き込みます。

datasource0 = glueContext.create_dynamic_frame.from_options(connection_type="s3", connection_options = {"paths": [staging_path], "useS3ListImplementation":True,"recurse":True}, format="json")
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [map_spec])
datasink2 = glueContext.write_dynamic_frame.from_options(frame = applymapping1, connection_type = "s3", connection_options = {"path": output_path}, format = "json")

AWS Glue コンソールでプロファイルされたメトリクスを可視化する

次のダッシュボードでは、入力ジョブから取得された Amazon S3 バイトの書き込みメトリクスを、出力ジョブの同じタイムラインにある Amazon S3 バイトの読み取りメトリクスに重ね合わせています。タイムラインには、入力および出力ジョブのさまざまなジョブの実行が表示されます。入力ジョブ (赤で表示) は、30 分ごとに開始されます。出力ジョブ (茶色で表示) は、入力ジョブの完了時に開始し、最大同時実行数は 1 です。

この例では、ジョブのブックマークが有効になっていません。スクリプトコードでジョブのブックマークを有効にするために使用される変換コンテキストはありません。

ジョブ履歴: [履歴] タブに示されているように、入力および出力ジョブには、午後 12 時に始まる複数の実行があります。

AWS Glue コンソールで入力ジョブは次のようになります。

以下のイメージに、出力ジョブを示します。

最初のジョブの実行: 以下のデータバイトの読み書きグラフに示されているように、12:00 から 12:30 の間に行われた入力および出力ジョブの最初のジョブの実行は、曲線の下のほぼ同じ領域を示しています。これらの領域は、入力ジョブによって書き込まれた Amazon S3 バイトと、出力ジョブによって読み取られた Amazon S3 バイトを表しています。このデータは、書き込まれた Amazon S3 バイトの比率によっても確認されます (30 分間の累計 – 入力ジョブのジョブトリガー頻度)。午後 12 時に開始された入力ジョブの実行の比率のデータポイントも 1 です。

以下のグラフは、すべてのジョブの実行におけるデータフロー比率を示しています。

2 回目のジョブの実行: 2 回目のジョブの実行では、入力ジョブによって書き込まれるバイト数と比較して、出力ジョブにより読み取られるバイト数に明確な違いがあります。(出力ジョブの 2 つのジョブの実行にまたがる曲線の下の領域を比較するか、入力および出力ジョブの 2 回目の実行の領域を比較します)。読み取りバイトと書き込みバイトの比率は、12:30 から 13:00 までの 2 番目の 30 分スパンにおいて入力ジョブによって書き込まれるデータの約 2.5 倍のデータを出力ジョブが読み取ることを示しています。ジョブのブックマークが有効になっていなかったため、出力ジョブが入力上部による最初のジョブの実行の出力を再処理したことがその理由です。1 より大きい比率は、出力ジョブにより処理された追加のデータバックログがあることを示しています。

3 回目のジョブの実行: 入力ジョブは、書き込まれるバイト数の観点でほぼ一定です (赤色の曲線の下にある領域を参照)。ただし、入力ジョブの 3 回目の実行には、予想されるよりも時間がかかります (赤色の曲線が長く続いていることを確認)。その結果、出力ジョブの 3 回目のジョブの実行が遅れて開始します。3 回目のジョブの実行では、13:00 から 13:30 までの残りの 30 分においてステージング場所に蓄積されたデータのごく一部のみ処理されました。バイトフローの比率は、入力ジョブの 3 回目の実行により書き込まれたデータの 0.83 のみ処理されたことを示しています (13:00 の比率を参照)。

入力ジョブと出力ジョブの重複: 入力ジョブの 4 回目のジョブの実行は、スケジュールに従って 13:30 に開始されました。これは、出力ジョブの 3 回目のジョブの実行が終了する前です。これらの 2 つのジョブの実行は一部の重複しています。ただし、出力ジョブの 3 回目の実行では、13:17 前後に開始されたとき、Amazon S3 のステージング場所にリストされたファイルのみキャプチャされます。これは、入力ジョブの最初のジョブの実行から取得されたすべてのデータ出力で構成されています。13:30 時点での実際の比率は約 2.75 です。出力ジョブの 3 回目のジョブの実行では、13:30 から 14:00 に入力ジョブの 4 回目のジョブの実行により書き込まれたデータの約 2.75 倍のデータが処理されました。

これらのイメージが示すように、出力ジョブは、入力ジョブのそれ以前のすべてのジョブの実行から取得されたステージング場所からのデータを再処理します。その結果、出力ジョブの 4 番目のジョブの実行が最も長くなり、入力ジョブの 5 回目のジョブの実行全体と重複します。

ファイルの処理を修正する

出力ジョブが、出力ジョブの以前のジョブの実行で処理されていないファイルのみ処理するようにする必要があります。これを行うには、次のように、ジョブのブックマークを有効にし、ジョブ出力に変換コンテキストを設定します。

datasource0 = glueContext.create_dynamic_frame.from_options(connection_type="s3", connection_options = {"paths": [staging_path], "useS3ListImplementation":True,"recurse":True}, format="json", transformation_ctx = "bookmark_ctx")

ジョブのブックマークが有効な場合、出力ジョブは、入力ジョブのそれまでのすべてのジョブの実行から取得されたステージング場所のデータを再処理しません。読み書きされたデータを示す以下のイメージでは、茶色の曲線の下にある領域はほぼ一定であり、赤色の曲線と似ています。

追加のデータが処理されないため、バイトフローの比率もほぼ 1 に維持されます。

出力ジョブのジョブの実行が開始され、次の入力ジョブの実行がデータをさらにステージング場所に置き始める前にステージング場所のファイルをキャプチャします。これを続けている限り、それまでの入力ジョブの実行からキャプチャされたファイルのみ処理され、比率はほぼ 1 に維持されます。

入力ジョブに予想以上に時間がかかるため、2 回の入力ジョブの実行から取得されたステージング場所のファイルが出力ジョブによってキャプチャされるとします。その出力ジョブの実行では、比率が 1 よりも大きくなります。ただし、出力ジョブのそれ以降のジョブの実行では、出力ジョブのそれまでのジョブの実行により既に処理されているファイルは処理されません。