GitHub から DAG をテスト、同期、デプロイする

Cloud Composer 1 | Cloud Composer 2

このガイドでは、CI / CD パイプラインを作成して、DAG のテスト、同期、および GitHub リポジトリから Cloud Composer 環境へのデプロイを行う方法について説明します。

他のサービスのデータのみを同期する場合は、他のサービスからデータを転送するをご覧ください。

CI / CD パイプラインの概要

図 1.フロー手順を示す構成図（クリックして拡大）

DAG をテスト、同期、デプロイする CI/CD パイプラインには、次のステップがあります。

1. DAG に変更を加えて、その変更をリポジトリ内の開発ブランチに push します。

2. リポジトリのメインブランチに対して pull リクエストを開きます。

3. Cloud Build で単体テストが行われ、DAG が有効であることを確認します。

4. pull リクエストが承認され、リポジトリのメインブランチに統合されます。

5. Cloud Build で新しい Cloud Composer 開発環境とこれらの新しい変更が同期されます。

6. DAG が開発環境で想定どおりに動作していることを確認します。

7. DAG が想定どおりに機能する場合は、DAG を本番環境の Cloud Composer にアップロードします。

目標

目的:

• Cloud Build を使用して自動送信前チェックを実行します。このチェックでは、DAG の単体テストが実行されます。
• 開発環境の Cloud Composer の DAG を GitHub リポジトリの DAG と同期します。

始める前に

• このガイドでは、開発環境と本番環境という 2 つの同一の Cloud Composer 環境を使用していることを前提としています。

  このガイドでは、開発環境のみに CI/CD パイプラインを構成します。使用する環境が本番環境でないことを確認します。

• このガイドでは、DAG とそのテストが GitHub リポジトリに保存されていることを前提としています。

  サンプルの CI/CD パイプラインに、サンプル リポジトリの内容が示されています。DAG とテストは dags/ ファイルに保存され、要件ファイル、制約ファイル、Cloud Build 構成ファイルが最上位に保存されます。DAG 同期ユーティリティとその要件は utils ディレクトリにあります。

  この構造は、Airflow 1、Airflow 2、Cloud Composer 1、Cloud Composer 2 の各環境で使用できます。

送信前チェックジョブと単体テストを作成する

最初の Cloud Build ジョブは、送信前チェックを実行し、DAG の単体テストを実行します。

単体テストを追加する

DAG の単体テストをまだ作成していない場合は作成します。これらのテストは、リポジトリ内の DAG と一緒に _test 接尾辞を付けて保存します。たとえば、example_dag.py の DAG のテストファイルは example_dag_test.py です。これらは、リポジトリで送信前チェックとして実行されるテストです。

送信前チェック用の Cloud Build YAML 構成を作成する

リポジトリで、test-dags.cloudbuild.yaml という名前の YAML ファイルを作成します。このファイルは、送信前チェック用に Cloud Build ジョブを構成します。ここには 3 つのステップがあります。

1. DAG に必要な依存関係をインストールします。
2. 単体テストに必要な依存関係をインストールします。
3. DAG テストを実行します。

steps:
  # install dependencies
  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements.txt", "-c", "constraints.txt", "--user"]

  - name: python:3.8-slim
    entrypoint: pip
    args: ["install", "-r", "requirements-test.txt", "--user"]

  # run in python 3.8 which is latest version in Cloud Composer
  - name: python:3.8-slim
    entrypoint: python3.8
    args: ["-m", "pytest", "-s", "dags/"]

注: Cloud Build は Docker Hub から Python イメージを pull します。タグは、各ステップの name フィールドで指定できます。entrypoint として python を使用している場合は、entrypoint がタグで指定されたバージョンと一致していることを確認してください。それ以外の場合は、entrypoint はデフォルトで latest になります。

送信前チェック用の Cloud Build トリガーを作成する

GitHub からのリポジトリのビルドガイドに沿って、次の構成で GitHub アプリベースのトリガーを作成します。

• 名前: test-dags

• イベント: pull リクエスト

• ソース - リポジトリ: 自分のリポジトリを選択

• ソース - ベースブランチ: ^main$（main は必要に応じてリポジトリのベースブランチの名前に変更します）

• ソース - コメント制御: 不要

  注意: コメント制御を必須から外すと、どのユーザーもビルドをトリガーする pull リクエストを送信できます。大規模なリポジトリや、信頼できないコントリビューターがいるリポジトリの場合は、コメント制御をおすすめします。

• ビルド構成 - Cloud Build 構成ファイル: /test-dags.cloudbuild.yaml（ビルドファイルのパス）

DAG 同期ジョブを作成して DAG ユーティリティ スクリプトを追加する

次に、DAG ユーティリティ スクリプトを実行する Cloud Build ジョブを構成します。このジョブのユーティリティ スクリプトは、リポジトリのメインブランチにマージされた DAG を Cloud Composer 環境と同期します。

注: Cloud Composer では、繰り返しすべての DAG が解析されます。これは、前回の解析以降に DAG が変更されていなくても発生します。このため、このユーティリティは変更されたものだけでなく、すべての DAG を Cloud Composer にアップロードします。

DAG ユーティリティ スクリプトを追加する

DAG ユーティリティ スクリプトをリポジトリに追加します。このユーティリティ スクリプトは、すべての非 DAG Paython ファイルを無視して、リポジトリの dags/ ディレクトリにあるすべての DAG ファイルを一時ディレクトリにコピーします。次に、スクリプトは、Cloud Storage クライアント ライブラリを使用して、この一時ディレクトリにあるすべてのファイルを Cloud Composer 環境のバケットの dags/ ディレクトリにアップロードします。

from __future__ import annotations

import argparse
import glob
import os
from shutil import copytree, ignore_patterns
import tempfile

# Imports the Google Cloud client library
from google.cloud import storage

def _create_dags_list(dags_directory: str) -> tuple[str, list[str]]:
    temp_dir = tempfile.mkdtemp()

    # ignore non-DAG Python files
    files_to_ignore = ignore_patterns("__init__.py", "*_test.py")

    # Copy everything but the ignored files to a temp directory
    copytree(dags_directory, f"{temp_dir}/", ignore=files_to_ignore, dirs_exist_ok=True)

    # The only Python files left in our temp directory are DAG files
    # so we can exclude all non Python files
    dags = glob.glob(f"{temp_dir}/*.py")
    return (temp_dir, dags)

def upload_dags_to_composer(
    dags_directory: str, bucket_name: str, name_replacement: str = "dags/"
) -> None:
    """
    Given a directory, this function moves all DAG files from that directory
    to a temporary directory, then uploads all contents of the temporary directory
    to a given cloud storage bucket
    Args:
        dags_directory (str): a fully qualified path to a directory that contains a "dags/" subdirectory
        bucket_name (str): the GCS bucket of the Cloud Composer environment to upload DAGs to
        name_replacement (str, optional): the name of the "dags/" subdirectory that will be used when constructing the temporary directory path name Defaults to "dags/".
    """
    temp_dir, dags = _create_dags_list(dags_directory)

    if len(dags) > 0:
        # Note - the GCS client library does not currently support batch requests on uploads
        # if you have a large number of files, consider using
        # the Python subprocess module to run gsutil -m cp -r on your dags
        # See https://cloud.google.com/storage/docs/gsutil/commands/cp for more info
        storage_client = storage.Client()
        bucket = storage_client.bucket(bucket_name)

        for dag in dags:
            # Remove path to temp dir
            dag = dag.replace(f"{temp_dir}/", name_replacement)

            try:
                # Upload to your bucket
                blob = bucket.blob(dag)
                blob.upload_from_filename(dag)
                print(f"File {dag} uploaded to {bucket_name}/{dag}.")
            except FileNotFoundError:
                current_directory = os.listdir()
                print(
                    f"{name_replacement} directory not found in {current_directory}, you may need to override the default value of name_replacement to point to a relative directory"
                )
                raise

    else:
        print("No DAGs to upload.")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--dags_directory",
        help="Relative path to the source directory containing your DAGs",
    )
    parser.add_argument(
        "--dags_bucket",
        help="Name of the DAGs bucket of your Composer environment without the gs:// prefix",
    )

    args = parser.parse_args()

    upload_dags_to_composer(args.dags_directory, args.dags_bucket)

DAG を同期するための Cloud Build YAML 構成を作成する

リポジトリで、add-dags-to-composer.cloudbuild.yaml という名前の YAML ファイルを作成します。このファイルは、DAG を同期するために Cloud Build ジョブを構成します。このファイルについては、以下の 2 つのステップがあります。

1. DAG ユーティリティ スクリプトに必要な依存関係をインストールします。

2. ユーティリティ スクリプトを実行して、リポジトリ内の DAG を Cloud Composer 環境と同期します。

steps:
  # install dependencies
  - name: python
    entrypoint: pip
    args: ["install", "-r", "utils/requirements.txt", "--user"]

  # run
  - name: python
    entrypoint: python
    args: ["utils/add_dags_to_composer.py", "--dags_directory=${_DAGS_DIRECTORY}", "--dags_bucket=${_DAGS_BUCKET}"]

注: Cloud Build は Docker Hub から Python イメージを pull します。タグは、各ステップの name フィールドで指定できます。entrypoint として python を使用している場合は、entrypoint がタグで指定されたバージョンと一致していることを確認してください。それ以外の場合は、entrypoint はデフォルトで latest になります。

Cloud Build トリガーを作成する

GitHub からのリポジトリのビルドガイドに沿って、次の構成で GitHub アプリベースのトリガーを作成します。

• 名前: add-dags-to-composer

• イベント: ブランチに push する

• ソース - リポジトリ: 自分のリポジトリを選択

• ソース - ベースブランチ: ^main$（main は必要に応じてリポジトリのベースブランチの名前に変更します）

• ソース - 含まれるファイル フィルタ（glob）: dags/**

• ビルド構成 - Cloud Build 構成ファイル: /add-dags-to-composer.cloudbuild.yaml（ビルドファイルのパス）

詳細設定で 2 つの置換変数を追加します。

• _DAGS_DIRECTORY - リポジトリに DAG が配置されているディレクトリ。 このガイドのサンプル リポジトリを使用している場合は、dags/ です。

• _DAGS_BUCKET - 開発環境の Cloud Composer の dags/ ディレクトリを含む Cloud Storage バケット。gs:// 接頭辞を省略します。例: us-central1-example-env-1234ab56-bucket

注意: _DAGS_BUCKET 置換変数の Cloud Storage バケット名から dags/ 接頭辞を削除しない場合の場合、DAG 同期ジョブは NotFound エラーで失敗します。これは、Cloud Storage では dags/ がフォルダであり、バケット名の一部ではなくオブジェクト名の一部と見なされるためです。

CI/CD パイプラインをテストする

このセクションでは、新しく作成された Cloud Build トリガーを利用する DAG 開発フローに従います。

送信前ジョブを実行する

メインブランチへの pull リクエストを作成して、ビルドをテストします。ページで送信前チェックを見つけます。[詳細] をクリックして [Google Cloud Build の詳細を表示する] を選択し、Google Cloud コンソールでビルドログを表示します。

重要: Cloud Build は、requirements.txt と requirements-test.txt で指定されたパッケージで単体テストを実行します。これには、Airflow の公開バージョンが含まれます。Cloud Composer 環境によっては、PyPI パッケージとプリインストールされたパッケージのセットが異なる場合があります。DAG の開発中は、ローカルの Airflow 環境を実行することをおすすめします。

図 2. GitHub での Cloud Build に関する送信前チェック ステータスのスクリーンショット（クリックして拡大）

送信前チェックが失敗した場合は、ビルドの失敗に対処するをご覧ください。

DAG が開発環境の Cloud Composer で動作することを確認する

pull リクエストが承認された後、メインブランチに統合します。Google Cloud コンソールを使用してビルド結果を表示します。Cloud Build トリガーが多数ある場合は、トリガー名 add-dags-to-composer でビルドをフィルタリングできます。

Cloud Build 同期ジョブが成功すると、同期された DAG が開発環境の Cloud Composer に表示されます。そこで、DAG が想定どおりに機能していることを確認できます。

DAG を本番環境にデプロイします。

DAG が想定どおりに動作するようになったら、本番環境に手動で追加します。これを行うには、Cloud Composer 環境のバケット内の dags/ ディレクトリに DAG ファイルをアップロードします。

DAG 同期ジョブが失敗した場合、または開発環境の Cloud Composer で DAG が想定どおりに動作しない場合は、ビルドの失敗に対処するをご覧ください。

ビルドの失敗に対処する

このセクションでは、一般的なビルドエラーのシナリオに対処する方法について説明します。

送信前チェックが失敗した場合はどうなりますか？

pull リクエストで [詳細] をクリックし、[Google Cloud Build の詳細を表示する] を選択して Google Cloud Console にビルドログを表示します。これらのログを使用して、DAG の問題をデバッグします。問題を解決したら、修正を commit してブランチに push します。送信前チェックが再実行され、デバッグツールとしてログを引き続き使用できます。

DAG 同期ジョブが失敗した場合はどうなりますか？

Google Cloud コンソールを使用してビルド結果を表示します。Cloud Build トリガーが多数ある場合は、トリガー名 add-dags-to-composer でビルドをフィルタリングできます。ビルドジョブのログを調べて、エラーを解決します。エラーの解決にサポートが必要な場合は、サポート チャネルをご利用ください。

DAG が Cloud Composer 環境で適切に機能しない場合はどうなりますか？

DAG が開発環境の Cloud Composer で想定どおりに機能しない場合は、DAG を本番環境の Cloud Composer に手動で昇格しないでください。代わりに、次のいずれかを行います。

• DAG を破損した変更を使用して pull リクエストを元に戻し、変更の直前の状態に復元する（これにより、pull リクエスト内の他のすべてのファイルも元に戻されます）
• 破損した DAG の変更を手動で元に戻すには、新しい pull リクエストを作成します。
• 新しい pull リクエストを作成して DAG のエラーを修正します。

これらの手順のいずれかを実行すると、新しい送信前チェックがトリガーされ、統合時に DAG 同期ジョブがトリガーされます。

次のステップ

• ローカルの Airflow 環境を実行する
• DAG を書き込む
• DAG をトリガーする
• DAG をテストする