[Airflow] DB 쿼리의 결과를 감시하고, 특정 조건이 충족될 때까지 작업을 일시 중지하는 기능 | SqlSensor

Apache Airflow의 SqlSensor는 데이터베이스 쿼리의 결과를 감시하고, 특정 조건이 충족될 때까지 작업을 일시 중지하는 역할을 합니다. 이 센서는 주로 데이터베이스 쿼리의 결과를 확인하여 특정 값이나 조건이 충족되었는지를 확인하는 데 사용됩니다.

 

SqlSensor 사용 방법

from airflow.sensors.sql import SqlSensor

sql_sensor_task = SqlSensor(
    task_id='sql_sensor_task',
    conn_id='your_database_connection_id',  # 데이터베이스 연결 ID
    sql='SELECT COUNT(*) FROM your_table WHERE your_condition;',  # 감시할 쿼리
    mode='poke',  # 'poke' 또는 'reschedule' 중 선택
    timeout=600,  # 타임아웃 (초 단위)
    poke_interval=60,  # 'poke' 모드에서 쿼리 감시 간격 (초 단위)
    dag=dag  # Airflow DAG 객체
)


# DAG에 추가 
sql_sensor_task >> your_next_task

• conn_id: 데이터베이스 연결을 식별하는 문자열. 이 연결은 Airflow의 Connection에서 설정해야 한다
• sql: 감시할 데이터베이스 쿼리
• mode: 센서 동작 모드. 'poke'는 지속적으로 감시하고, 'reschedule'은 일정 간격으로 새로운 작업을 예약
• timeout: 센서 작업을 중지할 최대 시간 (초 단위).
• poke_interval: 'poke' 모드에서 쿼리를 다시 실행할 간격 (초 단위).

SqlSensor는 설정된 데이터베이스 쿼리의 결과를 주기적으로 감시하며, 특정 조건이 충족되면 다음 작업이 실행한다.

 

 

데이터베이스 연결 정보 설정

Airflow에서 데이터베이스 연결 정보는 Airflow의 Connection에서 설정할 수 있는데, 이는 Airflow의 메타데이터베이스와 관련이 있으며 DAG (Directed Acyclic Graph) 실행 동안 사용된다.

Airflow 웹 인터페이스 또는 Airflow 명령행 인터페이스를 사용하여 Connection을 설정할 수 있고, 다음은 웹 인터페이스를 이용한 connection 설정 방법이다.

1. Airflow 웹 인터페이스로 이동
2. Admin 메뉴에서 Connections을 선택
3. "Create" 버튼을 클릭하여 새로운 Connection을 추가
4. 연결에 필요한 정보 (호스트, 포트, 사용자 이름, 암호 등)를 입력하고 저장