알루 코딩

이제까지는 상위 task가 하나 있을 때, 상위 task가 하위 task로 분기하는 경우 분기하는 조건에 대해서 알아봤습니다. 이번에는 상위 task 여러개가 하나의 하위 task로 연결이 되는 구조에서 하위 task의 실행조건을 설정하는 방법(Trigger Rule)에 대해 알아보겠습니다. 1. Trigger Rule 종류 아래와 같은 task들이 있다고 해봅시다. 기본적으로는 task1, task2, task3이 모두 성공적으로 잘 끝나야 task4가 도는 구조입니다. 하지만, 이 rule을 바꾸고 싶을 경우가 있을 것입니다. 예를 들어, task1, task2, task3 중에서 어느 하나라도 정상적으로 끝나면, task4가 수행되도록 하는 rule을 만들 수도 있고, 성공여부에 상관없이 task1..

이번 포스팅에서는 Task 분기처리하는 방법 중에 마지막 방법인 BaseBranchOperator로 분기처리하는 방법에 대해 살펴보겠습니다. 코드를 바로 살펴보겠습니다. from airflow.operators.branch import BaseBranchOperator with DAG(... ) as dag: class CustomBranchOperator(BaseBranchOperator): def choose_branch(self, context): import random item_lst = ['A','B','C'] selected_item = random.choice(item_lst) if selected_item == 'A': return 'task_a' elif selected_item in ..

이번 포스팅에서는 데커레이터인 task.branch 를 이용해서 task 분기처리를 해보려고 합니다. 코드를 바로 살펴보겠습니다. from airflow.decorators import task @task.branch(task_id='python_branch_task') def select_random(): import random item_lst = ['A','B','C'] selected_item = random.choice(item_lst) if selected_item == 'A': return 'task a' elif selected_item in ['B','C']: return ['task_b','task_c'] select_random() >> [task_a,task_b,task_c] Bra..

이제부터는 Task를 다룰 수 있는 고급기능에 대해 알아볼건데요, 그 중에서 먼저 Task를 분기처리할 수 있는 방법들에 대해살펴보도록 하겠습니다. Task 분기 처리하는 방법에는 크게 3가지가 있는데, 이번 포스팅에서는 BranchPython Operator에 대해 배워보겠습니다. 1. Task 분기 처리 유형 먼저, Task 분기처리는 왜 필요할까요? 예를 들어, 상위 task 1개에 하위 task 3개가 있다고 가정해봅시다. 기본적으로, 위 사진처럼 task관계가 있다면, task1이 끝나고, task2-1, task2-2, task2-3 가 모두 다 같이 수행됩니다. 그런데, 가끔은 task1의 수행결과에 따라서 task2-x 중 하나만 수행하도록 구성하고 싶은 경우가 있을 것입니다. 예를 들어,..

앞서서는 특정 dag에 있는 task끼리만 데이터를 공유할 수 있는 방법이었다면, 모든 dag에서 데이터를 접근하는 방식에 대해 알아볼까합니다. 그런 용도로 airflow에서 기능을 제공하는 것이 Variable 입니다. 1. 전역변수 Variable 이해 Xcom: 특정 DAG, 특정 schedule에 수행되는 Task 간에만 공유 모든 DAG이 공유할 수 있는 전역 변수는 없을까요? → 바로 Variable이 있습니다! ※ Variable 등록하기 : airflow 서비스를 띄우고, Admin 탭에 들어가서 Variables 메뉴를 누르고 "+" 버튼을 누르면 됩니다. : 실제 Variable의 Key, Value 값은 메타 DB에 저장됩니다. (variable 테이블) Variable도 Key, V..