Airflow 데이터베이스에 액세스

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

이 페이지에서는 Cloud Composer 환경의 Airflow 데이터베이스를 실행하는 Cloud SQL 인스턴스에 연결하고 SQL 쿼리를 실행하는 방법을 설명합니다.

예를 들어 Airflow 데이터베이스에서 직접 쿼리를 실행하거나, 데이터베이스 백업을 수행하거나, 데이터베이스 콘텐츠를 기반으로 통계를 수집하거나, 데이터베이스에서 다른 커스텀 정보를 가져올 수 있습니다.

시작하기 전에

Airflow 데이터베이스에서 SQL 쿼리 실행

Airflow 데이터베이스에 연결하려면 다음 단계를 따르세요.

  1. 하나 이상의 PostgresOperator 연산자로 DAG를 만듭니다. 시작하려면 예시 DAG를 사용하면 됩니다.

  2. 연산자의 sql 매개변수에서 SQL 쿼리를 지정합니다.

  3. 이 DAG를 환경에 업로드합니다.

  4. DAG를 트리거합니다. 예를 들어 수동으로 트리거하거나 예약된 시간에 실행될 때까지 기다릴 수 있습니다.

DAG 예시:

import datetime
import os

import airflow
from airflow.providers.postgres.operators.postgres import PostgresOperator

SQL_DATABASE = os.environ["SQL_DATABASE"]

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

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

데이터베이스 콘텐츠를 덤프하고 버킷으로 전송

다음 단계