Monkeykim's blog

default_args란 무엇인가?default_args는 DAG 하위의 모든 오퍼레이터에 공통으로 적용할 파라미터를 정의하는 데 사용됩니다. 이 파라미터는 반복적으로 동일한 값을 작성해야 하는 번거로움을 줄여주고, 코드의 일관성을 유지하는 데 기여합니다.default_args에 어떤 파라미터를 설정할 수 있을까?default_args에 설정할 수 있는 파라미터는 Airflow의 BaseOperator 클래스 생성자에서 제공하는 파라미터들입니다.예를 들어, 다음과 같은 파라미터들이 대표적입니다:start_dateend_dateretriesretry_delayon_failure_callbackon_success_callback기타 BaseOperator에서 제공되는 다양한 파라미터들Airflow Base ..

Airflow를 사용하다 보면 DAG 간의 의존성을 관리해야 하는 상황이 자주 발생합니다. 특히 Trigger DAG Run Operator나 External Task Sensor를 사용해 DAG 간 연결을 만들다 보면, 강한 연결 구조로 인해 관리가 복잡해지고 확장성이 떨어지는 문제가 생길 수 있습니다. 이런 문제를 해결하기 위해 Airflow는 Dataset 기능을 제공합니다.Dataset의 필요성기존 문제점: 강한 연결 구조DAG 간 의존성을 Trigger Operator와 External Sensor로 처리하면, 연결 관리에 많은 노력이 필요합니다. 이러한 방식은 강한 연결(Strong Coupling) 구조를 형성하기 때문에, DAG가 많아질수록 확장성과 유연성이 떨어집니다.해결책: 약한 연결 ..

DAG 간 의존성을 설정하는 방법Airflow에서는 DAG 간 의존성을 설정하는 데 두 가지 주요 방법을 제공합니다.1. TriggerDagRunOperator한 DAG의 특정 작업이 완료되었을 때 다른 DAG을 실행하도록 설정할 수 있습니다.DAG 간의 순차적 실행이 필요할 때 유용합니다.2. ExternalTaskSensor외부 DAG의 특정 작업(Task)이 완료된 후 현재 DAG의 작업(Task)을 실행하고 싶을 때 사용됩니다.작업 간의 동기화를 보장하며, 복잡한 DAG 네트워크에서 의존성을 쉽게 관리할 수 있습니다.https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/sensors/external_task/index.html ai..

File Sensor란?Airflow의 FileSensor는 파일 시스템에서 특정 파일이나 디렉토리가 존재하는지 감지하기 위해 사용됩니다. 워크플로우에서 다른 작업이 파일의 존재 여부에 의존할 때 매우 유용합니다. FileSensor는 주기적으로 파일을 확인하며, 파일이 존재하면 성공으로 처리되고, 존재하지 않으면 설정된 간격 동안 재시도합니다.FileSensor의 주요 매개변수fs_conn_id (File Connection ID)Airflow에서 설정된 파일 연결 ID를 참조합니다.기본적으로 FileSystemHook을 통해 연결을 관리합니다.filepath (파일 경로)파일의 상대 경로를 지정합니다.fs_conn_id로 설정된 기본 경로를 기준으로 상대적인 위치를 입력합니다.recursive (재귀..

Sensor Operator란?Sensor Operator는 특정 조건이 충족될 때까지 대기하는 Task로, 조건이 만족되면 True를 반환하여 다음 Task로 넘어갑니다. 모든 Sensor는 BaseSensorOperator를 상속하여 구현되며, 반드시 두 가지 메서드를 오버라이딩해야 합니다:__init__(): 초기화 메서드로, Sensor의 기본 설정을 정의합니다.poke(context): 센싱 로직을 정의하는 메서드로, 특정 조건을 만족하는지 확인합니다.BaseSensorOperator는 Airflow의 BaseOperator를 상속하며, Sensor의 동작과 모드를 결정하는 다양한 매개변수를 제공합니다.Airflow docs: https://airflow.apache.org/docs/apache..

Airflow에서 사용 가능한 Providers는 UI의 "Connections" 섹션에서 확인할 수 있습니다. 만약 필요한 Provider가 없다면 설치해야 합니다.설치 방법1. 도커 파일 생성Base Image로 사용중이던 airflow version의 이미지를 입력합니다.원하는 provider 패키지를 입력해줍니다. 2. 도커 이미지 빌드sudo docker build -t airflow_custom . 3. docker-compose.yaml 수정 airflow 서비스의 이미지를 airflow_custom으로 변경합니다. 4. 컨테이너 재실행docker-compose downdocker-compose up 5. Provider 확인Airflow UI의 Admin > Providers에서 설치된 P..

이번 포스팅에서는 Docker Compose로 PostgreSQL 컨테이너를 설정하고, Airflow의 Hook을 활용해 데이터베이스 작업을 간소화하는 방법을 살펴보겠습니다.PostgreSQL 컨테이너 설정Docker Compose를 활용해 PostgreSQL 컨테이너를 설정합니다. 아래는 docker-compose.yaml 파일에서 PostgreSQL 설정의 주요 부분입니다.services: postgres_custom: image: postgres:13 environment: POSTGRES_USER: kim POSTGRES_PASSWORD: kim POSTGRES_DB: kim TZ: Asia/Seoul volumes: - postgre..

Docker Compose의 역할과 사용 방법Docker Compose의 목적Docker Compose는 다중 컨테이너를 생성하고 관리할 때 효율성을 제공합니다. 예를 들어, 데이터베이스, 웹서버, 백엔드 등 여러 구성 요소를 하나의 애플리케이션으로 묶어야 할 때 유용합니다. Compose 파일 하나에 모든 컨테이너의 설정을 정의하고, 한 번의 명령어로 일괄적으로 컨테이너를 관리할 수 있습니다.Docker Compose 작성 방법Docker Compose 설정 파일은 docker-compose.yaml이라는 이름으로 작성합니다. 이 파일에 각 컨테이너의 이미지, 환경 변수, 볼륨, 네트워크, 의존성 등을 YAML 형식으로 정의할 수 있습니다.Docker Compose 서비스 시작 및 중지설정 파일이 있는..

1. DAG 파일 로드 및 파싱Airflow에서 모든 워크플로는 DAG(Directed Acyclic Graph) 형태로 정의됩니다. 이 DAG은 Python 코드로 작성되며, 작업의 순서와 의존성을 설정하는 방향성 비순환 그래프입니다. Airflow의 스케줄러(Scheduler)는 설정된 디렉토리를 주기적으로 확인하여 새로 생성되거나 수정된 DAG 파일이 있는지 감지합니다.(이로 인해 코드를 작성할 때 메모리 효율적으로 작성을 해야 함)스케줄러는 DAG 파일을 로드하고 파싱하여 DAG 객체를 생성하며, 이를 메타데이터 데이터베이스에 저장합니다. 이때 DAG에 정의된 모든 작업(Task)과 그 의존 관계도 함께 등록됩니다. 덕분에 Airflow는 DAG의 실행 시점을 정확히 파악하고 DAG이 완료될 때까..

Custom Operator가 필요한 이유기본적으로 Airflow에는 PythonOperator, BashOperator, EmailOperator 등 자주 사용하는 작업에 대한 Operator가 준비되어 있지만, 모든 상황을 커버하지는 않습니다. 예를 들어, 외부 API 호출 후 데이터를 처리하는 작업이나 복잡한 데이터 파이프라인을 위한 연산이 필요할 때는 Custom Operator가 더욱 적합합니다.Custom Operator를 통해 우리는 다음과 같은 이점을 얻을 수 있습니다.코드 재사용성: 특정 비즈니스 로직을 담은 Operator를 재사용하여 일관성 있게 워크플로우를 구축할 수 있습니다.코드 간소화: DAG에서 반복적인 코드를 줄이고, 명확한 역할을 가진 Operator를 만들어 가독성을 높일..