CS

Airflow Dynamic Task Mapping Dynamic Task Mapping — Airflow Documentation airflow.apache.org Astronomer Create dynamic Airflow tasks Create dynamic Airflow tasks | Astronomer Documentation How to dynamically create tasks at runtime in your Airflow DAGs. docs.astronomer.io 기본 매핑 매핑으로 task 생성하기 반복 매핑하기 매핑하지 않을 인수 입력하기 여러 인수 매핑하기 TaskFlow가 아닌 경우의 매핑 기존 오퍼레이터의 반환 값으로 매핑하기 TaskFlow와 기존 오퍼레이터 함께 사용하기 ..

Params — Airflow Documentation airflow.apache.org DAG 레벨 param Task 레벨 param Task에서 param 참조하기 JSON 스키마 검증 param을 통해 트리거 UI 폼 만들기 런타임 param 전달 끄기 Param param을 통하면 task에 런타임으로 구성을 전달할 수 있습니다. - 1. DAG 코드에 DAG 레벨 기본 param 지정 - 2. task 레벨로 param을 지정해 DAG 레벨 param 덮어쓰기 - 3. DAG 트리거 시(manually) param 전달 - 3 > 2 > 1 순으로 우선권을 갖습니다. - param은 JSON 스키마로 검증됩니다. - param을 잘 정의하면, UI에서 DAG 트리거 시 param 지정을 편리하..

XCom XComs — Airflow Documentation airflow.apache.org XCom (cross-communication)은 task 간 데이터를 전달할 수 있는 메커니즘입니다. Airflow는 여러 기기에 걸쳐 실행될 수 있고, 이때 task는 다른 task와 완전히 독립적으로 실행되기 때문에 이런 메커니즘을 사용합니다. XCom은 key (본질적으로 해당 XCom의 이름)와 해당 XCom이 발생한 task_id 및 dag_id로 식별합니다. XCom은 (직렬화된) 값을 가지는데, 원칙적으로 작은 데이터만 가능합니다. 데이터프레임 같은 큰 데이터를 전달하는데 사용하지 않는 게 좋습니다. XCom은 오퍼레이터나 task instance 클래스의 xcom_push와 xcom_pull을..

Celery Executor — Airflow Documentation airflow.apache.org 원격 executor에는 CeleryExecutor, KubernetesExecutor, DaskExecutor, CeleryKubernetesExecutor, LocalKubernetesExecutor가 있는데, 이중 세팅이 간단한 CeleryExecutor만 알아보도록 하겠습니다. Celery Executor 구조 Task 실행 프로세스 대기열 Celery Executor CeleryExecutor는 worker를 scale out할 수 있는 방법 중 하나입니다. Celery 시스템을 사용하여 scheduler 등 Airflow의 프로세스와 여러 worker를, 대기열을 사용하여 연결하는 방식입니..

Executor — Airflow Documentation airflow.apache.org Executor 개요 Executor 종류 로컬 Executor dag.test()로 DAG 테스트 LocalExecutor SequentialExecutor Executor 개요 Executor는 task instance가 무엇으로 실행될지에 관한 메커니즘입니다. executor는 일반적인 API를 가지며 "pluggable"하므로 필요에 따라 executor를 변경할 수 있습니다. Airflow는 한 번에 한 executor만 다룰 수 있습니다. 이는 구성 파일의 [core]에서 executor 옵션으로 설정할 수 있습니다. 내장 executor는 이름을 기재하면 됩니다. [core] executor = Ku..

Airflow는 여러 개의 기기를 사용하는 이른바 multi node로도 사용할 수 있습니다. 보통 scheduler 등을 실행하는 기기 1대와 worker를 실행하는 여러 기기로 구성합니다. Airflow는 이러한 클러스터를 쉽게 구성할 수 있는 Celery 시스템을 사용하는 executor인 Celery Executor를 지원합니다. Celery executor를 사용해 클러스터를 구성해보도록 하겠습니다. 노드 2개를 만들어서 하나에서 scheduler, webserver, celery flower를 실행하고, 다른 하나는 worker를 실행하도록 하겠습니다. 가상 환경 만들기 WSL 2개 설치 컴퓨터 1대를 노드 2개로 사용하기 위하여 WSL를 두 개 사용하였습니다. Microsoft Store에서..

Airflow TaskFlow Context Logging 인수로 임의 객체 전달 사용자 정의 객체 버전 관리 센서 DAG를 오퍼레이터보다 순수 파이썬 코드로 채울 생각이라면, @task 데코레이터를 쓰는 TaskFlow API가 도움이 됩니다. 또한 PythonOperator를 쓰는 것보다는 TaskFlow를 쓰는 게 코드를 줄이고 간결하게 쓸 수 있습니다. TaskFlow는 Task 간 XCom을 통한 입력과 출력을 처리하고, 의존성을 자동적으로 설정합니다. DAG 파일에서 TaskFlow 함수가 호출되면, 결과 값을 나타내는 XCom(XComArg)가 생성되고, 이를 downstream task 또는 오퍼레이터에 입력 값으로 전달합니다. from airflow.decorators import tas..

Airflow Operators, Airflow Sensors OperatorSensorJinja 템플릿파이썬 내장 객체로 렌더링하기 Operator오퍼레이터는 사전에 정의된 Task 템플릿입니다. DAG에서 오퍼레이터를 선언하여 task를 만들 수 있습니다.※ Task와 오퍼레이터는 대부분 상호 교환 가능한 개념입니다. 다만 Task를 말할 때는 DAG의 "실행 단위"를 뜻하고, 오퍼레이터를 말할 때는 재사용 가능한 미리 만들어진 task 템플릿을 가리킵니다.with DAG("my-dag") as dag: ping = SimpleHttpOperator(endpoint="http://example.com/update/") email = EmailOperator(to="admin@example...