1. はじめに
Apache Airflow は、定期的なスケジュールで DAG を実行するように設計されていますが、Cloud Storage バケットの変更や Cloud Pub/Sub に push されたメッセージなどのイベントに応答して DAG をトリガーすることもできます。これを行うには、Cloud Functions によって Managed Apache Airflow DAG をトリガーします。
このラボの例では、Cloud Storage バケットで変更が生じるたびに簡単な DAG を実行します。この DAG は、BashOperator を使用して、Cloud Storage バケットにアップロードされた変更情報を出力する bash コマンドを実行します。
このラボを開始する前に、マネージド Apache Airflow の概要と Cloud Functions スタートガイドの Codelab を完了しておくことをおすすめします。Managed Apache Airflow の概要 Codelab で Managed Apache Airflow 環境を作成した場合は、このラボでその環境を使用できます。
作成するアプリの概要
この Codelab では、次のことを行います。
- Google Cloud Storage にファイルをアップロードします。
- Node.JS ランタイムを使用して Google Cloud Functions をトリガーする
- この関数は、Google Cloud Managed Apache Airflow で DAG を実行します。
- これは、Google Cloud Storage バケットへの変更を出力する簡単な bash コマンドを実行します。
学習内容
- Google Cloud Functions と Node.js を使用して Apache Airflow DAG をトリガーする方法
必要なもの
- GCP アカウント
- JavaScript に関する基本的な知識
- Managed Apache Airflow/Airflow と Cloud Functions の基本的な知識
- CLI コマンドの使用に慣れている
2. GCP の設定
プロジェクトを選択または作成する
Google Cloud Platform プロジェクトを選択または作成します。新しいプロジェクトを作成する場合は、こちらの手順に沿って操作します。
プロジェクト ID をメモしておきます。この ID は後のステップで使用します。
新しいプロジェクトを作成する場合は、作成ページの [プロジェクト名] のすぐ下にプロジェクト ID が表示されます。 |
|
プロジェクトをすでに作成している場合は、コンソールのホームページの [プロジェクト情報] カードで ID を確認できます。 |
|
API を有効にする
|
Managed Apache Airflow 環境を作成する
次の構成で Managed Apache Airflow 環境を作成します。
他のすべての構成はデフォルトのままで問題ありません。下部の [作成] をクリックします。Managed Apache Airflow 環境の名前とロケーションをメモしておきます。これは、以降のステップで必要になります。 |
|
Cloud Storage バケットを作成する
次の構成で Cloud Storage バケットを作成します。
準備ができたら [作成] をクリックします。後の手順で使用するため、Cloud Storage バケットの名前をメモしておきます。 |
|
3. Google Cloud Functions（GCF）の設定
GCF を設定するには、Google Cloud Shell でコマンドを実行します。
Google Cloud は gcloud コマンドライン ツールを使用してノートパソコンからリモートで操作できますが、この Codelab では、Google Cloud Shell（Cloud 上で動作するコマンドライン環境）を使用します。
この Debian ベースの仮想マシンには、必要な開発ツールがすべて用意されています。永続的なホーム ディレクトリが 5 GB 用意されており、Google Cloud で稼働します。そのため、ネットワークのパフォーマンスと認証機能が大幅に向上しています。つまり、この Codelab に必要なのはブラウザだけです（Chromebook でも動作します）。
Google Cloud Shell を有効にするには、デベロッパー コンソールの右上にあるボタンをクリックします（環境のプロビジョニングと接続に若干時間を要します）。 |
|
Cloud Functions サービス アカウントに blob 署名権限を付与する
GCF が Airflow ウェブサーバーを保護するプロキシである Cloud IAP に対して認証を行うには、Appspot サービス アカウント GCF に Service Account Token Creator ロールを付与する必要があります。これを行うには、Cloud Shell で次のコマンドを実行します。<your-project-id> はプロジェクトの名前に置き換えます。
gcloud iam service-accounts add-iam-policy-binding \ <your-project-id>@appspot.gserviceaccount.com \ --member=serviceAccount:<your-project-id>@appspot.gserviceaccount.com \ --role=roles/iam.serviceAccountTokenCreator
たとえば、プロジェクトの名前が my-project の場合、コマンドは次のようになります。
gcloud iam service-accounts add-iam-policy-binding \ my-project@appspot.gserviceaccount.com \ --member=serviceAccount:my-project@appspot.gserviceaccount.com \ --role=roles/iam.serviceAccountTokenCreator
クライアント ID の取得
Cloud IAP に対する認証のためのトークンを作成するには、Airflow ウェブサーバーを保護するプロキシのクライアント ID が必要です。Managed Apache Airflow API はこの情報を直接提供しません。代わりに、認証されていないリクエストを Airflow ウェブサーバーに送信し、リダイレクト URL からクライアント ID を取得します。これを行うには、Cloud Shell を使用して Python ファイルを実行し、クライアント ID を取得します。
Cloud Shell で次のコマンドを実行して、GitHub から必要なコードをダウンロードします。
cd git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git
このディレクトリがすでに存在するためにエラーが発生した場合は、次のコマンドを実行して最新バージョンに更新します。
cd python-docs-samples/ git pull origin master
次のコマンドを実行して、適切なディレクトリに移動します。
cd python-docs-samples/composer/rest
Python コードを実行してクライアント ID を取得します。このとき、プロジェクトの名前を <your-project-id> に、前に作成した Managed Apache Airflow 環境のロケーションを <your-managed-airflow-location> に、前に作成した Managed Apache Airflow 環境の名前を <your-managed-airflow-environment> に置き換えます。
python3 get_client_id.py <your-project-id> <your-managed-airflow-location> <your-managed-airflow-environment>
たとえば、プロジェクト名が my-project、Managed Apache Airflow のロケーションが us-central1、環境名が my-airflow-environment の場合、コマンドは次のようになります。
python3 get_client_id.py my-project us-central1 my-airflow-environment
get_client_id.py によって、次の処理が行われます。
- Google Cloud で認証する
- リダイレクト URI を取得するために、認証されていない HTTP リクエストを Airflow ウェブサーバーに送信します。
- そのリダイレクトから
client_idクエリ パラメータを抽出する - 印刷して使用する
クライアント ID がコマンドラインに出力されます。次のような形式になります。
12345678987654321-abc1def3ghi5jkl7mno8pqr0.apps.googleusercontent.com
4. 関数を作成する
Cloud Shell で、次のコマンドを実行して、必要なサンプルコードを含むリポジトリのクローンを作成します。
cd git clone https://github.com/GoogleCloudPlatform/nodejs-docs-samples.git
必要なディレクトリに変更し、Cloud Shell を開いたまま、次の手順を完了します。
cd nodejs-docs-samples/composer/functions/composer-storage-trigger
ナビゲーション メニューをクリックし、[Cloud Functions] をクリックして、Google Cloud Functions ページに移動します。 |
|
ページ上部の [関数を作成] をクリックします。 |
|
関数に「my-function」という名前を付け、メモリはデフォルトの 256 MB のままにします。 |
|
[トリガー] を [Cloud Storage] に設定し、[イベントタイプ] を [Finalize/Create] のままにして、Cloud Storage バケットの作成の手順で作成したバケットを参照します。 |
|
[ソースコード] は [インライン エディタ] のままにして、ランタイムを [Node.js 8] に設定します。 |
|
Cloud Shell で次のコマンドを実行します。これにより、Cloud Shell エディタで index.js と package.json が開きます。
cloudshell edit index.js package.json
[package.json] タブをクリックしてコードをコピーし、Cloud Functions インライン エディタの [package.json] セクションに貼り付けます。 |
|
[実行する関数] を triggerDag に設定します。 |
|
[index.js] タブをクリックしてコードをコピーし、Cloud Functions インライン エディタの [index.js] セクションに貼り付けます。 |
|
|
|
Cloud Shell で次のコマンドを実行します。<your-environment-name> は Managed Apache Airflow 環境の名前に、<your-composer-region> は Managed Apache Airflow 環境が配置されているリージョンに置き換えます。
gcloud composer environments describe <your-environment-name> --location <your-managed-airflow-region>
たとえば、環境の名前が my-airflow-environment で、us-central1 にある場合、コマンドは次のようになります。
gcloud composer environments describe my-airflow-environment --location us-central1
出力結果は次のようになります。
config:
airflowUri: https://abc123efghi456k-tp.appspot.com
dagGcsPrefix: gs://narnia-north1-test-codelab-jklmno-bucket/dags
gkeCluster: projects/a-project/zones/narnia-north1-b/clusters/narnia-north1-test-codelab-jklmno-gke
nodeConfig:
diskSizeGb: 100
location: projects/a-project/zones/narnia-north1-b
machineType: projects/a-project/zones/narnia-north1-b/machineTypes/n1-standard-1
network: projects/a-project/global/networks/default
oauthScopes:
- https://www.googleapis.com/auth/cloud-platform
serviceAccount: 987665432-compute@developer.gserviceaccount.com
nodeCount: 3
softwareConfig:
imageVersion: composer-1.7.0-airflow-1.10.0
pythonVersion: '2'
createTime: '2019-05-29T09:41:27.919Z'
name: projects/a-project/locations/narnia-north1/environments/my-airflow-environment
state: RUNNING
updateTime: '2019-05-29T09:56:29.969Z'
uuid: 123456-7890-9876-543-210123456
出力で、 |
|
[その他] プルダウン リンクをクリックし、地理的に最も近いリージョンを選択します。 |
|
[失敗時に再試行する] をオンにする |
|
[作成] をクリックして Cloud Functions の関数を作成します |
|
コードのステップ実行
index.js からコピーしたコードは次のようになります。
// [START composer_trigger]
'use strict';
const fetch = require('node-fetch');
const FormData = require('form-data');
/**
* Triggered from a message on a Cloud Storage bucket.
*
* IAP authorization based on:
* https://stackoverflow.com/questions/45787676/how-to-authenticate-google-cloud-functions-for-access-to-secure-app-engine-endpo
* and
* https://cloud.google.com/iap/docs/authentication-howto
*
* @param {!Object} data The Cloud Functions event data.
* @returns {Promise}
*/
exports.triggerDag = async data => {
// Fill in your Managed Apache Airflow environment information here.
// The project that holds your function
const PROJECT_ID = 'your-project-id';
// Navigate to your webserver's login page and get this from the URL
const CLIENT_ID = 'your-iap-client-id';
// This should be part of your webserver's URL:
// {tenant-project-id}.appspot.com
const WEBSERVER_ID = 'your-tenant-project-id';
// The name of the DAG you wish to trigger
const DAG_NAME = 'composer_sample_trigger_response_dag';
// Other constants
const WEBSERVER_URL = `https://${WEBSERVER_ID}.appspot.com/api/experimental/dags/${DAG_NAME}/dag_runs`;
const USER_AGENT = 'gcf-event-trigger';
const BODY = {conf: JSON.stringify(data)};
// Make the request
try {
const iap = await authorizeIap(CLIENT_ID, PROJECT_ID, USER_AGENT);
return makeIapPostRequest(
WEBSERVER_URL,
BODY,
iap.idToken,
USER_AGENT,
iap.jwt
);
} catch (err) {
throw new Error(err);
}
};
/**
* @param {string} clientId The client id associated with the Managed Apache Airflow webserver application.
* @param {string} projectId The id for the project containing the Cloud Function.
* @param {string} userAgent The user agent string which will be provided with the webserver request.
*/
const authorizeIap = async (clientId, projectId, userAgent) => {
const SERVICE_ACCOUNT = `${projectId}@appspot.gserviceaccount.com`;
const JWT_HEADER = Buffer.from(
JSON.stringify({alg: 'RS256', typ: 'JWT'})
).toString('base64');
let jwt = '';
let jwtClaimset = '';
// Obtain an Oauth2 access token for the appspot service account
const res = await fetch(
`http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/${SERVICE_ACCOUNT}/token`,
{
headers: {'User-Agent': userAgent, 'Metadata-Flavor': 'Google'},
}
);
const tokenResponse = await res.json();
if (tokenResponse.error) {
return Promise.reject(tokenResponse.error);
}
const accessToken = tokenResponse.access_token;
const iat = Math.floor(new Date().getTime() / 1000);
const claims = {
iss: SERVICE_ACCOUNT,
aud: 'https://www.googleapis.com/oauth2/v4/token',
iat: iat,
exp: iat + 60,
target_audience: clientId,
};
jwtClaimset = Buffer.from(JSON.stringify(claims)).toString('base64');
const toSign = [JWT_HEADER, jwtClaimset].join('.');
const blob = await fetch(
`https://iam.googleapis.com/v1/projects/${projectId}/serviceAccounts/${SERVICE_ACCOUNT}:signBlob`,
{
method: 'POST',
body: JSON.stringify({
bytesToSign: Buffer.from(toSign).toString('base64'),
}),
headers: {
'User-Agent': userAgent,
Authorization: `Bearer ${accessToken}`,
},
}
);
const blobJson = await blob.json();
if (blobJson.error) {
return Promise.reject(blobJson.error);
}
// Request service account signature on header and claimset
const jwtSignature = blobJson.signature;
jwt = [JWT_HEADER, jwtClaimset, jwtSignature].join('.');
const form = new FormData();
form.append('grant_type', 'urn:ietf:params:oauth:grant-type:jwt-bearer');
form.append('assertion', jwt);
const token = await fetch('https://www.googleapis.com/oauth2/v4/token', {
method: 'POST',
body: form,
});
const tokenJson = await token.json();
if (tokenJson.error) {
return Promise.reject(tokenJson.error);
}
return {
jwt: jwt,
idToken: tokenJson.id_token,
};
};
/**
* @param {string} url The url that the post request targets.
* @param {string} body The body of the post request.
* @param {string} idToken Bearer token used to authorize the iap request.
* @param {string} userAgent The user agent to identify the requester.
*/
const makeIapPostRequest = async (url, body, idToken, userAgent) => {
const res = await fetch(url, {
method: 'POST',
headers: {
'User-Agent': userAgent,
Authorization: `Bearer ${idToken}`,
},
body: JSON.stringify(body),
});
if (!res.ok) {
const err = await res.text();
throw new Error(err);
}
};
// [END composer_trigger]
状況を確認してみましょう。ここでは、triggerDag、authorizeIap、makeIapPostRequest の 3 つの関数があります。
triggerDag は、指定された Cloud Storage バケットに何かをアップロードしたときにトリガーされる関数です。PROJECT_ID、CLIENT_ID、WEBSERVER_ID、DAG_NAME などの他のリクエストで使用される重要な変数を構成します。authorizeIap と makeIapPostRequest を呼び出します。
exports.triggerDag = async data => {
// Fill in your Managed Apache Airflow environment information here.
// The project that holds your function
const PROJECT_ID = 'your-project-id';
// Navigate to your webserver's login page and get this from the URL
const CLIENT_ID = 'your-iap-client-id';
// This should be part of your webserver's URL:
// {tenant-project-id}.appspot.com
const WEBSERVER_ID = 'your-tenant-project-id';
// The name of the DAG you wish to trigger
const DAG_NAME = 'composer_sample_trigger_response_dag';
// Other constants
const WEBSERVER_URL = `https://${WEBSERVER_ID}.appspot.com/api/experimental/dags/${DAG_NAME}/dag_runs`;
const USER_AGENT = 'gcf-event-trigger';
const BODY = {conf: JSON.stringify(data)};
// Make the request
try {
const iap = await authorizeIap(CLIENT_ID, PROJECT_ID, USER_AGENT);
return makeIapPostRequest(
WEBSERVER_URL,
BODY,
iap.idToken,
USER_AGENT,
iap.jwt
);
} catch (err) {
throw new Error(err);
}
};
authorizeIap は、サービス アカウントを使用して Airflow ウェブサーバーを保護するプロキシにリクエストを行い、makeIapPostRequest の認証に使用される ID トークンと JWT を「交換」します。
const authorizeIap = async (clientId, projectId, userAgent) => {
const SERVICE_ACCOUNT = `${projectId}@appspot.gserviceaccount.com`;
const JWT_HEADER = Buffer.from(
JSON.stringify({alg: 'RS256', typ: 'JWT'})
).toString('base64');
let jwt = '';
let jwtClaimset = '';
// Obtain an Oauth2 access token for the appspot service account
const res = await fetch(
`http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/${SERVICE_ACCOUNT}/token`,
{
headers: {'User-Agent': userAgent, 'Metadata-Flavor': 'Google'},
}
);
const tokenResponse = await res.json();
if (tokenResponse.error) {
return Promise.reject(tokenResponse.error);
}
const accessToken = tokenResponse.access_token;
const iat = Math.floor(new Date().getTime() / 1000);
const claims = {
iss: SERVICE_ACCOUNT,
aud: 'https://www.googleapis.com/oauth2/v4/token',
iat: iat,
exp: iat + 60,
target_audience: clientId,
};
jwtClaimset = Buffer.from(JSON.stringify(claims)).toString('base64');
const toSign = [JWT_HEADER, jwtClaimset].join('.');
const blob = await fetch(
`https://iam.googleapis.com/v1/projects/${projectId}/serviceAccounts/${SERVICE_ACCOUNT}:signBlob`,
{
method: 'POST',
body: JSON.stringify({
bytesToSign: Buffer.from(toSign).toString('base64'),
}),
headers: {
'User-Agent': userAgent,
Authorization: `Bearer ${accessToken}`,
},
}
);
const blobJson = await blob.json();
if (blobJson.error) {
return Promise.reject(blobJson.error);
}
// Request service account signature on header and claimset
const jwtSignature = blobJson.signature;
jwt = [JWT_HEADER, jwtClaimset, jwtSignature].join('.');
const form = new FormData();
form.append('grant_type', 'urn:ietf:params:oauth:grant-type:jwt-bearer');
form.append('assertion', jwt);
const token = await fetch('https://www.googleapis.com/oauth2/v4/token', {
method: 'POST',
body: form,
});
const tokenJson = await token.json();
if (tokenJson.error) {
return Promise.reject(tokenJson.error);
}
return {
jwt: jwt,
idToken: tokenJson.id_token,
};
};
makeIapPostRequest は Airflow ウェブサーバーを呼び出して composer_sample_trigger_response_dag. をトリガーします。DAG 名は url パラメータで渡される Airflow ウェブサーバー URL に埋め込まれており、idToken は authorizeIap リクエストで取得したトークンです。
const makeIapPostRequest = async (url, body, idToken, userAgent) => {
const res = await fetch(url, {
method: 'POST',
headers: {
'User-Agent': userAgent,
Authorization: `Bearer ${idToken}`,
},
body: JSON.stringify(body),
});
if (!res.ok) {
const err = await res.text();
throw new Error(err);
}
};
5. DAG を設定する
Cloud Shell で、サンプル ワークフローを含むディレクトリに移動します。これは、クライアント ID の取得の手順で GitHub からダウンロードした python-docs-samples の一部です。
cd cd python-docs-samples/composer/workflows
Managed Apache Airflow に DAG をアップロードする
次のコマンドを使用して、サンプル DAG を Managed Apache Airflow 環境の DAG ストレージ バケットにアップロードします。ここで、<environment_name> は Managed Apache Airflow 環境の名前、<location> は環境が配置されているリージョンの名前です。trigger_response_dag.py は、使用する DAG です。
gcloud composer environments storage dags import \
--environment <environment_name> \
--location <location> \
--source trigger_response_dag.py
たとえば、Managed Apache Airflow 環境の名前が my-airflow-environment で、us-central1 にある場合、コマンドは次のようになります。
gcloud composer environments storage dags import \
--environment my-airflow-environment \
--location us-central1 \
--source trigger_response_dag.py
DAG のステップ実行
trigger_response.py の DAG コードは次のようになります。
import datetime
import airflow
from airflow.operators import bash_operator
default_args = {
'owner': 'Managed Apache Airflow Example',
'depends_on_past': False,
'email': [''],
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': datetime.timedelta(minutes=5),
'start_date': datetime.datetime(2017, 1, 1),
}
with airflow.DAG(
'composer_sample_trigger_response_dag',
default_args=default_args,
# Not scheduled, trigger only
schedule_interval=None) as dag:
# Print the dag_run's configuration, which includes information about the
# Cloud Storage object change.
print_gcs_info = bash_operator.BashOperator(
task_id='print_gcs_info', bash_command='echo {{ dag_run.conf }}')
default_args セクションには、Apache Airflow の BaseOperator モデルで必要なデフォルトの引数が含まれています。このセクションとこれらのパラメータは、任意の Apache Airflow DAG に表示されます。owner は現在 Managed Apache Airflow Example に設定されていますが、必要に応じて自分の名前に変更できます。depends_on_past は、この DAG が以前の DAG に依存していないことを示しています。3 つのメール セクション（email、email_on_failure、email_on_retry）は、この DAG のステータスに基づいてメール通知が届かないように設定されています。retries が 1 に設定されているため、DAG は 1 回だけ再試行します。retry_delay に従って、5 分後に再試行します。通常、start_date は DAG を実行するタイミングを schedule_interval（後で設定）と組み合わせて指定しますが、この DAG の場合は関係ありません。2017 年 1 月 1 日に設定されていますが、過去の任意の日付に設定できます。
default_args = {
'owner': 'Managed Apache Airflow Example',
'depends_on_past': False,
'email': [''],
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': datetime.timedelta(minutes=5),
'start_date': datetime.datetime(2017, 1, 1),
}
with airflow.DAG セクションでは、実行する DAG を構成します。タスク ID composer_sample_trigger_response_dag、default_args セクションのデフォルト引数、最も重要な schedule_interval の None を使用して実行されます。この特定の DAG を Cloud Functions でトリガーするため、schedule_interval は None に設定されています。そのため、default_args の start_date は関連性がありません。
実行時に、DAG は print_gcs_info 変数で指定された構成を出力します。
with airflow.DAG(
'composer_sample_trigger_response_dag',
default_args=default_args,
# Not scheduled, trigger only
schedule_interval=None) as dag:
# Print the dag_run's configuration, which includes information about the
# Cloud Storage object change.
print_gcs_info = bash_operator.BashOperator(
task_id='print_gcs_info', bash_command='echo {{ dag_run.conf }}')
6. 関数をテストする
Managed Apache Airflow 環境を開き、環境名を含む行で Airflow リンクをクリックします。 |
|
名前をクリックして |
|
別のタブを開き、前に作成して Cloud Functions のトリガーとして指定した Cloud Storage バケットにファイルをアップロードします。これは、コンソールまたは gsutil コマンドを使用して行うことができます。 |
|
Airflow UI のタブに戻り、[グラフビュー] をクリックします。 |
|
|
|
メニューの右上にある [ログを表示] をクリックします。 |
|
ログには、Cloud Storage バケットにアップロードしたファイルに関する情報が表示されます。 |
|
おめでとうございます！これで、Node.js と Google Cloud Functions を使用して Airflow DAG をトリガーしました。
7. クリーンアップ
このクイックスタートで使用したリソースについて GCP アカウントに課金されないようにするには:
- （省略可）データを保存するには、Managed Apache Airflow 環境の Cloud Storage バケットと、このクイックスタート用に作成したストレージ バケットからデータをダウンロードします。
- 作成した環境の Cloud Storage バケットを削除します
- マネージド Apache Airflow 環境を削除します。環境を削除しても、環境の Storage バケットは削除されません。
- （省略可）サーバーレス コンピューティングでは、毎月最初の 200 万回の呼び出しは無料です。関数をゼロにスケーリングすると、課金されません（詳細については、料金をご覧ください）。ただし、Cloud Functions の関数を削除する場合は、関数の概要ページの右上にある [削除] をクリックします。
必要に応じて、プロジェクトを削除することもできます。
- GCP Console で、[プロジェクト] ページに移動します。
- プロジェクト リストで、削除するプロジェクトを選択し、[削除] をクリックします。
- ボックスにプロジェクト ID を入力し、[シャットダウン] をクリックしてプロジェクトを削除します。