Connections

호스트 이름, 포트 번호, 계정 정보 등 다른 시스템과 연결 관련 정보를 저장하는데 사용한다.

장점

소스 코드내에 민감한 정보를 하드코딩할 필요가 없다
관련 정보들을 한 군데서 관리할 수 있어서 편하다

사용하는 방법

Airflow 웹 페이지에서 메뉴바에서 [Admin] → [Connections]후 내용입력

---

airflow tasks

DAG에서 태스크들이 어떻게 실행될까?
Airflow 스케줄러는 DAG 단위가 아닌 태스크 단위로 스케줄을 한다

그럼 태스크는 왜 나눠야 할까?

태스크의 처리시간이 긴 경우 실패했을 때 재처리하는 시간이 길어진다. 해서 적절하게 나눠서 실패했을 때 재실행시간을 단축할 수 있도록 만들어야 한다

예를들어, 하나의 태스크를 처리하는데 1시간이 소요되는데 58분쯤에 실패가 된다면 다시 약 1시간의 시간을 들여 재실행해야 한다.

❗️태스크를 많이 나눴을 때 문제점은?

태스크를 많이 만들면 전체 DAG가 실행회는데 오래걸리고 스케줄러에 부하가 간다

Task Decorator 사용 3개의 태스크간 데이터 전달

태스크 간 주고 받는 데이터는 Airflow 메타데이터 DB에 기록됨
정확히는 Xcom이란 방식(테이블)을 사용함
DAG_RUN ID / task ID / key를 기본키로 저장한 후 다음 task에 꺼내 씀

Tasks test 제약

여기서 Tasks test 명령어 제약이 생기는데 기본적으로 Tasks test 명령어는 한 Tasks가 잘 실행되는지 확인을 전제로 사용하는 명령어라 Xcom으로 데이터를 받아서 테스트하는 태스크는 의미가 없다


예를들어, 아래와 같은 두개의 태스크가 있을 때 transform태스크만 airflow tasks test 명령어를 쓴다고 하더라고 위에서 data를 보내줄 태스크가 없기 때문에 의미가 없다


@task
def extract():
return {"order_id": 1}


@task
def transform(data):
print(data["order_id"])

그럼 이런 문제를 어떻게 해결할 수 있을까?

Xcom으로 데이터를 넘겨주기보단 파일경로를 하나 정의해두고 각 태스크마다 그 경로에서 파일을 확인하여 처리할 수 있도록 하면 된다


예를 들면, 파일명_YYYYMMDD_HHMMSS.csv의 형식으로(logical_date라는 명령을 활용)고정하여 각 태스크에서 맞는 파일명으로 보고 처리할 수 있도록 한다

기존

data
⬇️
@task
def task1(data)
	return data_t1
⬇️
@task
def task2(data_t1)
	return data_t2

개선

def get_file_path #파일 경로를 만들어둠

@task
def task1()
	file = get_file_path() # 경로 호출
    return 
    
@task
def task2()
	file = get_file_path() # 경로 호출
    return
    

task1에서 task2한테 전달하는 느낌이 아니라 하나의 경로를 정하고 함수가 그 경로를 호출하는 원리이다