[Airflow] Provider

아파치 에어플로우(Apache Airflow)의 프로바이더(Provider)는 Airflow가 다양한 외부 시스템과 상호작용할 수 있도록 기능을 확장하는 독립적인 설치 패키지입니다. Airflow 1.x 버전에서는 모든 외부 시스템 연동 기능(AWS, GCP, Slack, Postgres 등)이 Airflow 핵심 패키지 안에 내장되어 있었습니다. 이로 인해 패키지가 비대해지고, 특정 기능 하나를 업데이트하기 위해 Airflow 전체를 업그레이드해야 하는 불편함이 있었습니다.

Airflow 2.0부터 이러한 문제를 해결하기 위해 프로바이더 모델이 도입되었습니다. 이제 사용자는 필요한 외부 시스템의 Provider만 별도로 설치하여 Airflow의 기능을 가볍고 유연하게 확장할 수 있습니다. VScode/Pycharm의 plugin 같은 느낌이라고 할까요?

 

1. Provider의 구성요소

1) Hooks (훅)

외부 시스템과의 연결 및 인증을 담당하는 가장 기본적인 인터페이스입니다.
예를 들어, S3Hook은 AWS의 S3에, PostgresHook은 PostgreSQL 데이터베이스에 연결하는 기능을 제공합니다. API 클라이언트나 데이터베이스 드라이버를 감싸는 저수준(low-level)의 역할을 합니다.

 

2) Operators (오퍼레이터)

DAG 내에서 실제 작업을 수행하는 단위입니다. Operator는 내부적으로 Hook을 사용하여 외부 시스템과 통신합니다.
예를들어 S3ListOperator는 S3 버킷의 파일 목록을 조회하고, PostgresOperator는 특정 SQL 쿼리를 실행하는 작업을 정의합니다.

 

3) Sensors (센서)

Operator의 한 종류로, 특정 조건이 충족될 때까지 기다리는 역할을 합니다.
S3KeySensor는 특정 S3 버킷에 파일이 생길 때까지 기다리고, SqlSensor는 SQL 쿼리 결과가 특정 조건을 만족할 때까지 기다립니다.

 

4) Transfers (전송)

하나의 시스템에서 다른 시스템으로 데이터를 이동시키는 데 특화된 Operator입니다.
S3ToGoogleCloudStorageOperator는 AWS S3의 데이터를 GCP GCS로 복사하는 작업을 수행합니다.

 

 

2. Provider의 종류

Provider는 거의 모든 종류의 외부 시스템을 지원하며, 크게 다음과 같이 분류할 수 있습니다.

• 클라우드 플랫폼:
  • apache-airflow-providers-amazon: AWS S3, EC2, Redshift, EMR 등
  • apache-airflow-providers-google: GCP BigQuery, GCS, Dataflow 등
  • apache-airflow-providers-microsoft-azure: Azure Blob Storage, Data Lake 등
• 데이터베이스:
  • apache-airflow-providers-postgres
  • apache-airflow-providers-mysql
  • apache-airflow-providers-snowflake
• 서비스 및 도구:
  • apache-airflow-providers-slack: 슬랙으로 알림 메시지 전송
  • apache-airflow-providers-docker: 도커 컨테이너 실행
  • apache-airflow-providers-cncf-kubernetes: 쿠버네티스 Pod 실행
  • apache-airflow-providers-apache-spark: 스파크 잡(job) 제출

 

설치는 아래와 같이 간단히 할 수 있습니다. 

# Amazon Provider 설치 예시
pip install apache-airflow-providers-amazon

# Google Cloud와 PostgreSQL Provider 함께 설치
pip install apache-airflow-providers-google apache-airflow-providers-postgres

 

---

PostgreSQL과 연동하는 예시를 볼까요? 

PostgreSQL DB를 구성한 후 예시 테이블을 만들어줍니다. 사실 이것도 2가지 방법이 존재합니다.

---

1. 로컬 PostgreSQL Docker컨테이너에서 접근

만약 저처럼 로컬에 PostgreSQL이 구축되어 있고 Docker 컨테이너 안에서 airflow가 돌아가고 있다면, PostgreSQL에서 외부 연결을 받을 수 있도록 설정하는 것이 필요합니다. PostgreSQL에 설치된 디렉토리로 가서 (저의 경우는 D:\Program Files\PostgreSQL\17\data 안이었습니다. 만약 경로를 알고 싶으시면 Dbeaver같은 데 들어가서 아래와 같이 명령어를 치면 어디에 설치되어 있는지 알 수 있습니다)

SHOW config_file;

 

 

• PostgreSQL postgresql.conf에서 외부 IP를 허용:

listen_addresses = '*'

• pg_hba.conf에 접속 허용 규칙 추가 (예: 모든 IP 허용):

 

(저 같은 경우에는 원래 로컬(127.0.0.1, ::1)만 허용하고 있었습니다. 컨테이너에서 PostgreSQL로 접근하려면 외부 IP 대역을 허용해야 합니다.)

TYPE	DATABASE	USER	ADDRESS	METHOD

 

pg_hba.conf의 줄의 의미는 아래와 같습니다. 

 

• TYPE: local / host / hostssl / hostnossl
• DATABASE: 어떤 데이터베이스 허용할지 (all 가능)
• USER: 어떤 유저 허용할지 (all 가능)
• ADDRESS: 어떤 클라이언트 IP 허용할지 (CIDR 형식)
• METHOD: 인증 방식 (md5, scram-sha-256, trust 등)

 

# 모든 IPv4 주소 허용
host    all             all             0.0.0.0/0               scram-sha-256

 

혹은 도커 네트워크 대역만 허용한다면

host    all             all             172.18.0.0/16           scram-sha-256

 

도커 네트워크 대역을 확인하기위해서는 powershell에서

docker network ls

docker network inspect bridge

에서 Subnet이 도커 대역입니다. (bridge 네트워크가 기본 네트워크 입니다)

그리고 나면, PostgreSQL을 재시작해야합니다. 

sudo systemctl restart postgresql # linux

docker restart postgres # docker

# 먼저 정확한 서비스 이름을 확인합니다.
Get-Service postgres*

# 확인한 서비스 이름으로 재시작 명령을 실행합니다.
# 예: 서비스 이름이 'postgresql-x64-16'인 경우
Restart-Service -Name "postgresql-x64-16"

 

 

자, 이제 로컬 PostgreSQL에서의 설정은 마쳤으니 Docker 컨테이너에서 호스트로 접근해야합니다. 

 

로 세팅하고 test를 했을 때 초록색 " Connection successfully tested " 가 뜨면 잘 세팅된 겁니다. 

저는.. 여기서 여러가지 오류가 발생해서 시간을 정말 많이 소요했습니다. 

permission denied 
1) pg_hba.conf 설정
기본적으로 PostgreSQL은 localhost(127.0.0.1)에서만 접속을 허용하고, 외부 접속은 막혀 있을 수 있습니다. 
host    all    all    0.0.0.0/0    md5 와 같이 수정
2) Docker-desktop을 끄지 않아서 PostgreSQL 재시작 작동X
tasklist /FI "PID eq 21444" 라고 입력하면 http://m.docker.backend.exe 21444 Console

PostgreSQL 권한 부여 (혹시나해서..)
GRANT ALL PRIVILEGES ON DATABASE "FDATA" TO postgres;

 

참고) 아래처럼 직접 입력하는 방법도 있습니다. 

airflow connections add 'my_postgres' \
    --conn-type 'postgres' \
    --conn-host 'host.docker.internal' \  # or 호스트 IP
    --conn-login 'postgres' \
    --conn-password '비밀번호' \
    --conn-schema 'db이름' \
    --conn-port '5432'

 

 

2. Docker-compose이용 묶기

아니면 docker-compose.yml로 아예 Airflow와 PostgreSQL을 같은 네트워크로 묶는 방식도 생각해 볼 수 있습니다. 아래와 같은 장점이 있습니다. 

 

• 로컬 PostgreSQL 대신 Docker 안에서 관리하므로 환경이 더 깔끔해짐
• Airflow와 PostgreSQL이 같은 네트워크에 속해 있어서 localhost 대신 서비스 이름으로 접근 가능

version: "3"

services:
  postgres:
    image: postgres:14
    container_name: postgres
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow_db
    ports:
      - "5432:5432"   # 로컬에서 접근할 경우 필요
    networks:
      - airflow_net
    volumes:
      - postgres_data:/var/lib/postgresql/data

  airflow-webserver:
    image: apache/airflow:2.9.1
    container_name: airflow-webserver
    depends_on:
      - postgres
    environment:
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres:5432/airflow_db
    ports:
      - "8080:8080"
    command: webserver
    networks:
      - airflow_net

  airflow-scheduler:
    image: apache/airflow:2.9.1
    container_name: airflow-scheduler
    depends_on:
      - postgres
    environment:
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres:5432/airflow_db
    command: scheduler
    networks:
      - airflow_net

networks:
  airflow_net:

volumes:
  postgres_data:

 

예시에서는 airflow_net이라는 사용자 정의 네트워크를 만들었습니다. 

 

이제 Powershell에서 해당 경로로 이동 후 docker compose up해서 띄워줍시다.

 

---

 

그리고 Docker container의 airflow scheduler에 접속해서 pip install (아까 예시처럼) 명령어로 

pip install apache-airflow-providers-postgres 

와 같이 설치를 해줍니다. (사실 저 같은 경우에는 이미 airflow설치 단계에서 이미 같이 설치해둬서 이미 설치가 완료되었습니다)

설치 후에는 airflow standalone으로 재시작해주신 이후에 같은 방식으로 접속합니다. 

 

 

 

수많은 실패끝에 드디어 DAGs 실행해서 succes 확인!!