Cloud Composer 2 で Apache Airflow DAG を実行する
Cloud Composer 1 | Cloud Composer 2
このクイックスタート ガイドでは、Cloud Composer 環境を作成し、Cloud Composer 2 で Apache Airflow DAG を実行する方法について説明します。
Airflow を初めてご利用の場合は、Airflow のコンセプト、オブジェクト、使用状況の詳細について、Apache Airflow ドキュメントの Airflow のコンセプト チュートリアルをご覧ください。
代わりに Google Cloud CLI を使用する場合は、Cloud Composer で Apache Airflow DAG を実行する(Google Cloud CLI)をご覧ください。
Terraform を使用して環境を作成する場合は、環境の作成(Terraform)をご覧ください。
始める前に
- Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.
-
Google Cloud Console の [プロジェクト セレクタ] ページで、Google Cloud プロジェクトを選択または作成します。
-
Google Cloud Console の [プロジェクト セレクタ] ページで、Google Cloud プロジェクトを選択または作成します。
-
Enable the Cloud Composer API.
-
このクイックスタートを完了するために必要な権限を取得するには、プロジェクトに対する次の IAM ロールを付与するよう管理者に依頼してください。
-
Cloud Composer 環境を表示および管理するには:
-
環境とストレージ オブジェクトの管理者
(
roles/composer.environmentAndStorageObjectAdmin
) - サービス アカウント ユーザー(
roles/iam.serviceAccountUser
)
-
環境とストレージ オブジェクトの管理者
(
-
ログを表示するには: ログビューア (
roles/logging.viewer
)
ロールの付与の詳細については、アクセスの管理をご覧ください。
-
Cloud Composer 環境を表示および管理するには:
環境の作成
Google Cloud コンソールで、[環境の作成] ページに移動します。
プロジェクトの最初の環境の場合は、[Cloud Composer サービス アカウントに必要な権限を付与する] セクションが表示されます。
環境のサービス アカウントの新しいプリンシパルとして Cloud Composer サービス エージェント アカウントを追加し、Cloud Composer v2 API サービス エージェント拡張機能のロールを付与します。
環境で使用するサービス アカウントを使用していることを確認し、[付与] をクリックします。
[名前] フィールドに「
example-environment
」と入力します。[ロケーション] プルダウン リストで、Cloud Composer 環境のリージョンを選択します。このガイドでは、
us-central1
リージョンを使用します。その他の環境構成オプションには、指定されたデフォルト値を使用します。
[作成] をクリックして、環境が作成されるまで待ちます。
処理が完了すると、緑色のチェックマークが環境名の横に表示されます。
DAG ファイルを作成する
Airflow DAG は、スケジュールを設定して実行する体系的なタスクの集まりです。DAG は、標準の Python ファイルで定義されます。
このガイドでは、quickstart.py
ファイルで定義された Airflow DAG の例を使用します。このファイルの Python コードは、次のことを行います。
- DAG(
composer_sample_dag
)を作成します。この DAG は毎日実行されます。 - タスク(
print_dag_run_conf
)を実行します。このタスクは、bash 演算子を使用して DAG 実行の構成を出力します。
ローカルマシンに quickstart.py
ファイルのコピーを保存します。
DAG ファイルを環境のバケットにアップロードする
すべての Cloud Composer 環境には、Cloud Storage バケットが関連付けられています。Cloud Composer の Airflow は、このバケットの /dags
フォルダにある DAG のみをスケジュール設定します。
DAG のスケジュールを設定するには、quickstart.py
をローカルマシンから環境の /dags
フォルダにアップロードします:
Google Cloud コンソールで [環境] ページに移動します。
環境のリストで、ご利用の環境の名前、
example-environment
をクリックします。[環境の詳細] ページが開きます。[DAG フォルダを開く] をクリックします。[バケットの詳細] ページが開きます。
[ファイルをアップロード] をクリックして、
quickstart.py
のコピーを選択します。ファイルをアップロードするには、[開く] をクリックします。
DAG を表示する
DAG ファイルをアップロードすると、Airflow によって次の処理が行われます。
- アップロードした DAG ファイルを解析します。DAG が Airflow で使用可能になるまでに数分かかる場合があります。
- DAG を使用可能な DAG のリストに追加します。
- DAG ファイルで指定したスケジュールに従って DAG を実行します。
DAG UI で DAG を表示して、DAG がエラーなしで処理され、Airflow で使用できることを確認します。DAG UI は、Google Cloud コンソールで DAG 情報を表示するための Cloud Composer インターフェースです。Cloud Composer では、ネイティブの Airflow ウェブ インターフェースである Airflow UI にもアクセスできます。
以前にアップロードした DAG ファイルを Airflow が処理し、最初の DAG 実行(後述)を完了するまで、約 5 分間待ちます。
Google Cloud コンソールで [環境] ページに移動します。
環境のリストで、ご利用の環境の名前、
example-environment
をクリックします。[環境の詳細] ページが開きます。[DAG] タブに移動します。
composer_quickstart
DAG が DAG のリストに存在することを確認します。図 1.DAG のリストに composer_quickstart DAG が表示されます(クリックして拡大)
DAG 実行の詳細を表示する
DAG の 1 回の実行は DAG 実行と呼ばれます。DAG ファイルの開始日が昨日に設定されているため、Airflow はサンプルの DAG の DAG 実行をすぐに実行します。このようにして、Airflow は指定された DAG のスケジュールに追いつきます。
例の DAG には、コンソールで echo
コマンドを実行する 1 つのタスク print_dag_run_conf
が含まれています。このコマンドは、DAG に関するメタ情報(DAG 実行の数値識別子)を出力します。
[DAG] タブで、
composer_quickstart
をクリックします。DAG の [実行] タブが開きます。DAG 実行のリストで、最初のエントリをクリックします。
図 2.composer_quickstart DAG の DAG 実行のリスト(クリックして拡大) DAG 実行の詳細が表示され、サンプル DAG の個々のタスクに関する情報の詳細が表示されます。
図 3.DAG 実行で実行されたタスクのリスト(クリックして拡大) [DAG 実行のログ] セクションには、DAG 実行のすべてのタスクのログが一覧表示されます。ログで
echo
コマンドの出力を確認できます。図 4.print_dag_run_conf タスクのログ(クリックして拡大)
クリーンアップ
このページで使用したリソースについて、Google Cloud アカウントに課金されないようにするには、次の手順を行います。
このチュートリアルで使用したリソースを削除します。
Cloud Composer 環境を削除します。
Google Cloud Console で [環境] ページに移動します。
[
example-environment
] を選択し、[削除] をクリックします。環境が削除されるまで待ちます。
環境のバケットを削除します。Cloud Composer 環境を削除しても、バケットは削除されません。
Google Cloud Console で、[ストレージ] > [ブラウザ] ページに移動します。
環境のバケットを選択して、[削除] をクリックします。たとえば、このバケットの名前を
us-central1-example-environ-c1616fe8-bucket
にします。
環境の Redis のキューの永続ディスクを削除します。Cloud Composer 環境を削除しても、永続ディスクは削除されません。
Google Cloud Console で、[Compute Engine] > [ディスク] に移動します。
環境の Redis のキューの永続ディスクを選択し、[削除] をクリックします。
たとえば、このディスクの名前は
pvc-02bc4842-2312-4347-8519-d87bdcd31115
です。Cloud Composer 2 のディスクは常にBalanced persistent disk
タイプで、サイズ 2 GB です。
次のステップ