Airflow PostgreSQL 연결

airflow에 postgresql을 연결하는 법.
airflow에 connection을 추가해두면 host, port 등 정보를 코드에 노출하지 않고도 지정한 connection id만으로 연결이 가능하다.

Airflow UI를 통한 Connection 추가 방법

1. pip install 'apache-airflow-providers-postgres
   airflow 2.x 버전부터는 위처럼 pip install하면 된다.

2. 설치 후 스케줄러 실행
   airflow scheduler

3. 웹 UI 접속
   airflow webserver --port 8080

4. postgres provider 추가
   웹 'Admin' 탭 중 providers에 들어가면 다음과 같이 apache-airflow-providers-postgres provider가 추가되어 있다.
   
   

5. postgres connection 추가

• Admin 탭 중 connection 클릭 후 '+' 버튼 클릭
  
• Connection Type 중 Postgres 클릭
  

나머지는 본인의 postgresql 정보를 자유롭게 작성.

위 설정 후 코드에서 사용하는 법

1. postgres 연결

from airflow.hooks.postgres_hook import PostgresHook

rdb_hook = PostgresHook(postgres_conn_id='yoonene_postgres')
connection = rdb_hook.get_conn()

위처럼 PostgresHook에 지정했던 postgres_conn_id를 입력하면 연결이 된다.

2. query

ex)
import psycopg2.extras

cursor = connection.cursor(cursor_factory=psycopg2.extras.DictCursor)

query = f"SELECT * FROM {table_name}"
cursor.execute(query)
data = cursor.fetchall()

3. connection 종료

cursor.close()
connection.close()