알루 코딩

서울시 공공데이터 API 키를 발급받는 방법에 대해 소개해드리겠습니다. 아래 링크로 들어가셔서 로그인을 먼저 하시고 인증키 신청을 해야 합니다. https://data.seoul.go.kr/together/mypage/actkeyMain.do 열린데이터광장 메인 데이터분류,데이터검색,데이터활용 data.seoul.go.kr 실시간 지하철 오픈 API를 활용하려면 오른쪽의 '실시간 지하철 인증키 신청'을 누르고, 그 외의 데이터를 활용하려면 '일반 인증키 신청' 버튼을 클릭하시면 됩니다. 인증키를 신청할 때, 사용 URL에는 특별한 사이트 주소가 없다면 그냥 localhost라고 적으시고, 그 외의 칸도 각자의 이메일과 활용용도 등을 적으시면 됩니다. 다 기입하셨다면, 인증키 신청을 클릭합니다. 이제, 서..

이번에는 다른 DAG을 수행시킬 수 있는 'Trigger Run 오퍼레이터' 라는 것에 대해 알아보도록 하겠습니다. Airflow에서는 DAG간의 의존관계는 선-후행 관계입니다. 이를 설정할 수 있는 방법은 크게 2가지가 있는데요, 그 2개 중에 하나인, "TriggerDagRun" 오퍼레이터를 이번 포스팅에서 배워보겠습니다. 1. DAG 간 의존관계 설정 DAG 의존관계 설정 방법 (1) TriggerDagRun 오퍼레이터 오퍼레이터를 이용해서 Task를 만드는 것처럼 TriggerDagRun오퍼레이터로 Task를 만듭니다. task를 만들면서 파라미터를 줄 때, 어떤 DAG을 Trigger할 지 그 DAG의 id를 넣게 되어있습니다. 위 그림을 기준으로, task1이 선행 task고, task2, t..

이번 포스팅에서는 여러가지 airflow에서의 오퍼레이터로 뭐가 있는지 다양하게 살펴보겠습니다. 1. 기본 오퍼레이터 : Airflow에서 기본적으로 제공해주는 오퍼레이터를 한 번 살펴보도록 하겠습니다. 파일 경로 오퍼레이터(클래스) 중요도 비고 airflow.models.baseoperator BaseOperator ⭐⭐⭐ - 오펄레이터를 직접 개발하고 싶은 경우, 이 클래스 상속하여 개발 (execute() 함수를 오버라이딩 하여 사용합니다.) - 아래 오퍼레이터들은 모두 이 클래스를 상속하여 개발되어 있습니다. - Airflow를 잘 쓰려면, 이 오퍼레이터 상속/개발하는 것을 자유자재로 할 줄 알아야 합니다. airflow.operators.bash BashOperator ⭐⭐⭐ - bash쉘 스크..

오늘은 간단한 개념인 Edge Label을 짚고 넘어가볼까 합니다. 1. Edge Label 개념이란 무엇일까요? : Task 연결에 대한 설명(Comment)입니다. 굉장히 간단합니다. 말 그대로, Task 간의 edge에 있는 comment를 뜻합니다. 바로 실습을 해보도록 하죠. 2. Edge Label 실습 #1 from airflow.utils.edgemodifier import Label empty_1 = EmptyOperator( task_id='empty_1' ) empty_2 = EmptyOperator( task_id='empty_2' ) empty_1 >> Label('1과 2사이') >> empty_2 먼저, 라벨을 달려면, airflow.utils.edgemodifier에서 Lab..

이번 포스팅에서는 Task들을 모아서 관리할 수 있는 Task Group에 대해 알아보겠습니다. 하나의 dag에 task가 많다면, 관련있는 task끼리 그룹화하여 관리하도록 지원해주는 기능입니다. 1. Task Group 의 개념 Task들의 모음입니다. UI Graph 탭에서 Task들을 Group화하여 보여줍니다.(https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html#taskgroups) DAGs — Airflow Documentation airflow.apache.org 각 task가 그룹화된 섹션을 열어보면, inner_section이 또 있습니다. 즉, task 그룹 안에 또 다른 task 그룹도 계층적으로 ..

이제까지는 상위 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..