Airflow データベースにアクセスする

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

このページでは、Cloud Composer 環境の Airflow データベースを実行する Cloud SQL インスタンスに接続して SQL クエリを実行する方法について説明します。

たとえば、Airflow データベースで直接クエリを実行する、データベースのバックアップを作成する、データベースのコンテンツに基づいて統計情報を収集する、またはデータベースからその他のカスタム情報を取得することが必要になる場合があります。

始める前に

Airflow データベースで SQL クエリを実行する

Airflow データベースに接続するには:

  1. 1 つ以上の SQLExecuteQueryOperator 演算子を使用して DAG を作成します。始めるにあたっては、サンプル DAG を使用できます。

  2. 演算子の sql パラメータで、SQL クエリを指定します。

  3. この DAG を環境にアップロードします。

  4. DAG をトリガーします。たとえば、手動でトリガーすることも、スケジュールに従い実行されるまで待つこともできます。

DAG の例:

import datetime
import os

import airflow
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

SQL_DATABASE = os.environ["SQL_DATABASE"]

with airflow.DAG(
    "airflow_db_connection_example",
    start_date=datetime.datetime(2025, 1, 1),
    schedule_interval=None,
    catchup=False) as dag:

    SQLExecuteQueryOperator(
        task_id="run_airflow_db_query",
        dag=dag,
        conn_id="airflow_db",
        database=SQL_DATABASE,
        sql="SELECT * FROM dag LIMIT 10;",
    )

SQLExecuteQueryOperator の使用方法の詳細については、Airflow ドキュメントの SQLExecuteQueryOperator を使用した Postgres のハウツーガイドをご覧ください。

データベース コンテンツをダンプしてバケットに転送する

次のステップ