3장 실습에서는 파일 경로에 {{ data_interval_start }}를 넣었습니다. 그런데 그냥 datetime.now()로 현재 시각을 넣으면 안 될까요? 매시간 도는 DAG에서 10시가 조금 지나 시작한 실행은 10시가 아니라 9~10시의 데이터를 처리합니다. 그렇다면 이 "9시"는 어디서 받고, {{ ... }} 문자열과 Python 함수에는 언제 어떻게 채워질까요?
《Apache Airflow 기반의 데이터 파이프라인》 4장은 위키피디아의 시간별 페이지뷰를 가져오는 예제로 이 과정을 설명합니다. 날짜를 URL에 넣고, 필요한 페이지의 조회 수를 고른 뒤, 결과를 외부 데이터베이스에 저장합니다.
이 글에서 알아갈 수 있는 것
- 9~10시 데이터를 처리하는 실행에서
datetime.now()·logical_date·data_interval_start는 각각 몇 시일까? - 구간의 끝이 필요하면
logical_date에 한 시간을 더하면 될까? {{ ... }}는 언제 실제 값으로 바뀌고, 실행 전에 어떻게 확인할까?PythonOperator의 함수는 날짜를 객체로 받을까, 문자열로 받을까?op_args와op_kwargs는 어떻게 다를까?- 외부 DB에 접속할 때 Operator·Hook·Connection은 각각 무엇을 할까?
이번 장에서 연결할 핵심은 DAG를 정의할 때 작업의 틀을 만들고, 각 태스크를 실행할 때 그 실행에 맞는 값으로 틀을 채운다는 것입니다. 이때 값을 제공하는 묶음이 태스크 콘텍스트, 문자열의 빈칸을 채우는 도구가 Jinja 템플릿입니다.
1. 무엇을 만들까: 시간별 위키피디아 페이지뷰
책 4.1~4.1.1절 · 그림 4.1~4.2
예제의 배경은 가상의 주식시장 예측 도구 StockSense입니다. 어떤 기업의 위키피디아 페이지뷰가 늘면 그 기업에 대한 관심이 커져 주가도 오를 수 있다는 가정에서 출발합니다. 이번 장은 이 가정을 검증하지 않고, 검증에 쓸 Apple·Amazon·Google 같은 기업 페이지의 조회 수를 모으는 파이프라인까지 만듭니다.
위키미디어는 페이지뷰를 시간별 압축 파일로 공개합니다. UTC 기준 2019년 7월 17일 17~18시의 조회 수를 담은 파일 주소는 다음과 같습니다.
https://dumps.wikimedia.org/other/pageviews/2019/2019-07/pageviews-20190717-180000.gz
파일 이름의 180000은 집계 구간의 끝 시각입니다(그림 4.1). 그래서 18시 파일에는 17시부터 18시 전까지의 조회 수가 들어 있습니다. 이 점은 3절에서 URL을 만들 때 다시 씁니다.
URL의 연·월·일·시각을 바꾸면 다른 시간의 파일을 고를 수 있습니다. 3장에서 API의 조회 시작·종료 조건을 바꿨다면, 이번에는 이미 시간별로 나뉜 파일 중 하나를 선택합니다.
파일 안의 행은 다음 형태입니다.
en Apple 15 0
| 항목 | 예시 | 이번에 쓰는 방법 |
|---|---|---|
| 프로젝트 | en |
영어 위키피디아 행만 고릅니다. |
| 페이지 제목 | Apple |
관심 있는 페이지인지 고릅니다. |
| 조회 수 | 15 |
해당 시간의 조회 수로 저장합니다. |
| 응답 크기 | 0 |
위키미디어가 페이지뷰 파일에서는 집계를 빼서 0으로 나옵니다. 쓰지 않습니다. |
실제 시간별 파일에는 많은 페이지가 프로젝트·제목 순으로 정렬되어 있고, 같은 프로젝트·제목 쌍은 한 파일에 한 번만 나옵니다. 그래서 파일을 내려받은 뒤 필요한 행만 고르면 됩니다. 형식 설명은 위키미디어 페이지뷰 문서, 파일 목록은 페이지뷰 덤프에서 확인할 수 있습니다.
증분 데이터를 적재하는 방법 결정하기 (4.1.1)
필요한 시간마다 파일 하나를 고르면 됩니다. 실행할 때마다 모든 과거 파일을 다시 읽을 이유는 없습니다.
시간별 .gz 내려받기 → 압축 풀기 → 관심 페이지의 조회 수 추출 → 외부 DB에 저장
여기서 매번 바뀌는 것은 다운로드할 시간과 그 시간에 맞는 경로입니다. 날짜를 코드에 고정하면 실행 날짜가 바뀔 때마다 DAG 파일도 고쳐야 합니다. 4.2절의 템플릿은 이 값을 실행마다 채우기 위해 등장합니다.
실행한 시각과 맡은 시간은 다릅니다
그렇다면 "다운로드할 시간"은 무엇을 기준으로 정할까요? 가장 먼저 떠오르는 것은 datetime.now(), 즉 현재 시각입니다. 하지만 매시간 도는 DAG에서 9~10시 파일을 처리하는 실행은 10시가 지나서 시작합니다. 9~10시 데이터는 10시가 되어야 다 모이기 때문입니다. 3장에서 본 것처럼 Airflow는 구간이 끝난 뒤에 그 구간의 실행을 시작합니다.
그래서 한 실행에는 시각이 여러 개 붙습니다. 9~10시를 맡은 실행이라면 다음과 같습니다.
| 이름 | 값 | 뜻 |
|---|---|---|
data_interval_start |
09:00 | 이 실행이 맡은 구간의 시작 |
data_interval_end |
10:00 | 이 실행이 맡은 구간의 끝 |
logical_date |
09:00 | 실행에 붙는 날짜. 예약 실행에서는 구간의 시작과 같습니다 |
datetime.now() |
10:00 이후 어느 때 | 코드가 실제로 도는 순간의 시계 |
위의 세 값은 실행이 만들어질 때 정해지고, 언제 다시 돌려도 바뀌지 않습니다. datetime.now()만 실제로 돌린 때에 따라 달라집니다.
실습 컨테이너에 남은 실제 기록으로 비교해 보겠습니다(UTC). 결과 폴더 이름을 두 가지 방법으로 만든다고 생각해 봅니다.
| 맡은 구간 | 실제로 시작한 시각 | now()로 만든 폴더 |
구간 시작으로 만든 폴더 |
|---|---|---|---|
| 11월 2일 21~22시 (제때 실행) | 11월 2일 22:00 | 20261102-22 |
20261102-21 |
| 11월 2일 18~19시 (늦게 실행) | 11월 2일 19:50 | 20261102-19 |
20261102-18 |
| 11월 2일 21~22시 (일주일 뒤 다시 실행) | 11월 9일 05:37 | 20261109-05 |
20261102-21 |
- 첫 줄: 제때 돈 실행인데도
now()는 한 시간 뒤인 22시를 가리킵니다. - 둘째 줄: 시작이 50분 늦어지자
now()는 19시가 됩니다. - 셋째 줄: 첫 줄의 실행을 일주일 뒤에 다시 돌리자
now()는 11월 9일이 됩니다. 구간 시작으로 만든 폴더는 그대로라, 다시 돌린 결과가 원래의20261102-21폴더에 덮어써졌습니다.
언제 돌든 같은 구간은 같은 폴더에 들어가야 합니다. 그래서 처리할 시간은 현재 시각이 아니라 실행이 맡은 구간에서 꺼냅니다. 3절에서 이 값들이 들어 있는 콘텍스트를 자세히 봅니다.
2. 템플릿은 언제, 어디에서 채워질까?
책 4.2~4.2.1절 · 그림 4.3~4.4 · 리스트 4.1~4.2
오퍼레이터의 인수 템플릿 작업 (4.2.1)
Jinja는 문자열 안의 {{ ... }}를 값으로 바꾸는 템플릿 도구입니다. 예를 들어 BashOperator에 다음 명령을 넣을 수 있습니다.
BashOperator(
task_id="show_day",
bash_command="echo {{ ds }}",
)
DAG 파일을 읽고 이 오퍼레이터를 만들 때는 명령에 {{ ds }}가 남아 있습니다. 해당 태스크를 실행하기 전에, Airflow가 그 실행의 콘텍스트를 사용해 템플릿을 렌더링합니다. 렌더링은 템플릿을 최종 값으로 바꾸는 과정입니다.
논리적 날짜가 2026년 10월 24일인 실행이라면 위 명령은 다음처럼 됩니다.
echo 2026-10-24
그다음에 셸이 이 명령을 실행합니다. Jinja가 명령을 실행하는 것은 아닙니다.
이 순서 때문에 DAG 파일의 맨 위에서 datetime.now()로 날짜를 구하는 것과 차이가 납니다. 맨 위의 Python 코드는 DAG 파일을 파싱할 때 실행되므로, 그 값은 태스크가 맡은 구간과 상관없는 시각입니다. 1절의 표처럼 과거 구간을 다시 돌릴 때도 맡은 시간 대신 현재 시간을 쓰게 됩니다. 처리할 날짜는 실행 콘텍스트에서 받습니다.
모든 문자열이 템플릿은 아닙니다
중괄호를 적었다고 Airflow가 Python 파일 전체를 뒤져 바꾸지는 않습니다. 각 오퍼레이터가 template_fields에 등록한 속성이 렌더링 대상입니다.
아래 표는 오퍼레이터마다 {{ ... }}를 적을 수 있는 속성입니다. ds는 표에 없는데, 표에 있는 속성 안에 적어서 쓰는 값이기 때문입니다. 예를 들어 BashOperator의 bash_command 속성에 "echo {{ ds }}"처럼 적습니다. 실습 환경의 오퍼레이터에서 확인한 속성은 다음과 같습니다.
| 오퍼레이터 | 템플릿으로 처리하는 속성 |
|---|---|
BashOperator |
bash_command, env, cwd |
PythonOperator |
templates_dict, op_args, op_kwargs |
SQLExecuteQueryOperator |
sql, parameters, conn_id, database, hook_params |
따라서 bash_command="echo {{ ds }}"는 렌더링되지만, Python 함수 본문에 쓴 print("{{ ds }}")는 그대로 출력됩니다. 함수의 코드 자체는 PythonOperator의 템플릿 필드가 아니기 때문입니다.
또한 op_kwargs 안의 문자열처럼, 템플릿 필드에 들어 있는 사전·리스트의 값도 렌더링할 수 있습니다. 문자열을 어디에 넣었는지를 먼저 확인해야 합니다.
3. 실행 콘텍스트에는 어떤 값이 있을까?
책 4.2.2절 · 리스트 4.3~4.4 · 표 4.1
콘텍스트는 현재 태스크와 DAG 실행을 설명하는 값들의 묶음입니다. 날짜뿐 아니라 태스크 객체, DAG 실행 정보, 사용자가 준 매개변수도 들어 있습니다. Jinja 안에서는 이 이름을 바로 참조할 수 있습니다.
| 이름 | 의미 | 사용 예시 |
|---|---|---|
data_interval_start |
이번 실행이 맡은 데이터 구간의 시작 | URL에 넣을 시간 선택 |
data_interval_end |
이번 실행이 맡은 데이터 구간의 끝 | 조회 종료 조건 |
logical_date |
DAG 실행을 논리적으로 식별하는 날짜 | 실행 날짜 확인 |
ds |
논리적 날짜의 YYYY-MM-DD 문자열 |
일 단위 표시 |
ts |
논리적 날짜의 ISO 형식 문자열 | 시각을 포함한 표시 |
dag, task |
현재 DAG·태스크 객체 | DAG ID·태스크 ID 확인 |
ti, task_instance |
현재 태스크 인스턴스 | 태스크 실행 정보 확인 |
run_id |
DAG 실행 ID | 서로 다른 실행 구분 |
params |
사용자가 정의한 매개변수 | 처리할 페이지 목록 같은 설정 |
macros |
템플릿에서 쓸 수 있는 보조 함수·모듈 | 날짜 계산 |
var, conn |
Airflow Variable·Connection 접근 | 설정·연결 정보 조회 |
이 중 ds는 “오늘 날짜”라는 뜻이 아닙니다. 논리적 날짜를 문자열로 표시한 값입니다. 같은 날 여러 시간의 실행이 있으면 ds가 같을 수 있으니, 시간별 파일 이름으로는 부족합니다.
직접 실행하면 logical_date가 달라집니다
예약 실행에서는 logical_date가 구간의 시작과 같아서, ds로 날짜를 골라도 맞아 보입니다. 하지만 airflow dags test처럼 날짜를 지정해 직접 실행하면(수동 실행) logical_date는 지정한 그 시각이 됩니다. 데이터 구간은 그 시각 직전에 끝난 한 시간입니다.
- 10월 24일 10:00으로 실행하면:
logical_date는 10:00, 데이터 구간은 09:00~10:00입니다. - 10월 25일 00:30으로 실행하면:
logical_date는 25일 00:30이라ds는2026-10-25입니다. 그런데 데이터 구간은 24일 23:00~25일 00:00입니다.
두 번째 경우에 ds로 폴더를 만들면 24일 23시 데이터가 25일 폴더에 들어갑니다. 두 경우 모두 실습 컨테이너에서 위와 같은 값이 나왔습니다. 그래서 처리할 시간은 logical_date나 ds가 아니라 데이터 구간(data_interval_start·data_interval_end)에서 꺼냅니다.
데이터 구간에서 꺼내면 날짜가 넘어가도 따로 처리할 일이 없습니다. 마지막 실습의 DAG를 25일 00:30으로 실행해 보니 ds는 25일이었지만, 결과 폴더는 /tmp/ch04/20261024-23, 파일 주소는 pageviews-20261025-000000.gz였습니다. 둘 다 24일 23시~25일 00시 구간에 맞는 값입니다.
[Airflow 2.0 이후 달라진 점]
execution_date는 없어졌습니다4장 원본에는
execution_date.year·execution_date.hour와next_ds같은 이름이 나옵니다(표 4.1). 책의execution_date는 이름과 달리 "실행한 시각"이 아니라 구간의 시작이었습니다. 9~10시 구간의execution_date는 09:00이지만 실제 실행은 10시가 지나서 일어났습니다. Airflow 2.2에서data_interval_start·data_interval_end가 추가됐고, 공식 FAQ는execution_date를 예전 이름이라고 설명하면서 의미가 더 정확하고 오해가 적은 데이터 구간 이름을 쓰라고 권합니다.Airflow 3의 콘텍스트에는
execution_date와next_ds가 아예 없습니다. 3.3.2에서{{ execution_date.year }}를 렌더링하면'execution_date' is undefined오류가 납니다.
현재 사용 가능한 이름과 타입은 Airflow 3.3.2 템플릿 참조 문서에 정리되어 있습니다.
폴더는 구간의 시작, 파일 주소는 구간의 끝
콘텍스트의 날짜는 연·월·일·시각을 꺼내거나 형식을 바꿀 수 있는 객체입니다. 이번 예제는 같은 9~10시 구간에서 두 값을 씁니다.
| 쓰는 곳 | 쓰는 값 | 9~10시 구간의 결과 |
|---|---|---|
| 결과를 저장할 폴더 | data_interval_start |
20261024-09 |
| 내려받을 위키미디어 파일 | data_interval_end |
pageviews-20261024-100000.gz |
폴더는 "어느 시간의 결과인가"를 나타내므로 구간의 시작인 9시로 이름을 붙입니다.
{{ data_interval_start.in_timezone('UTC').strftime('%Y%m%d-%H') }}
파일 주소에는 구간의 끝을 씁니다. 1절에서 본 것처럼 위키미디어는 집계 구간이 끝나는 시각을 파일 이름에 붙이기 때문에, 9~10시 조회 수는 100000 파일에 들어 있습니다. 마침 Airflow도 9~10시 구간의 실행을 10시가 지나서 시작하고, 그 실행의 data_interval_end가 바로 10:00입니다. 그래서 구간의 끝을 그대로 파일 이름에 넣으면 됩니다.
pageviews-{{ data_interval_end.in_timezone('UTC').strftime('%Y%m%d-%H') }}0000.gz
책의 코드처럼 execution_date(Airflow 2의 예약 실행에서는 구간의 시작)로 고르면 090000.gz, 즉 8~9시 데이터를 받게 됩니다. 한 시간씩 어긋나는 셈입니다.
strftime('%H')는 시각을 두 자리로 채워 9시를 09로 적습니다. 8~9시 구간이라면 파일 이름이 pageviews-20261024-090000.gz여야 하는데, 시각 숫자 .hour를 그대로 붙이면 9가 들어가 pageviews-20261024-90000.gz라는 없는 이름이 됩니다.
실제로 내려받는다면, 파일은 보통 구간이 끝나고 45분 안에(늦으면 3~4시간 뒤) 공개되므로 재시도나 센서로 기다리게 해야 합니다.
구간의 끝은 직접 더하지 않습니다
앞에서 파일 주소에 구간의 끝을 썼습니다. 이 끝을 logical_date에 한 시간을 더해서 구해도 될까요? 템플릿에서는 macros.timedelta로 날짜를 더하거나 뺄 수 있어서, {{ logical_date + macros.timedelta(hours=1) }}처럼 쓸 수는 있습니다. 하지만 구간의 끝을 이렇게 직접 더하면 틀릴 수 있습니다.
- 이 DAG의 예약 실행: 09:00 + 1시간 = 10:00. 실제 구간의 끝과 같습니다.
- 10:00으로 직접 실행:
logical_date가 10:00이라 더하면 11:00이 됩니다. 실제 구간의 끝은 10:00입니다. - 평일 9시에만 도는 DAG에서
logical_date가 금요일 09:00인 예약 실행: 금요일 + 1일 = 토요일이지만, 이 실행이 맡은 구간은 금요일 09:00부터 다음 평일인 월요일 09:00까지입니다. 실행도 월요일 9시가 지나서 시작합니다.
구간의 끝은 우리가 계산하는 값이 아닙니다. Airflow가 실행을 만들 때 스케줄을 보고 계산해 data_interval_end에 넣어 두고, 우리는 꺼내 쓰기만 합니다. 직접 더해서 맞는 경우는 구간 길이가 늘 같고 logical_date가 구간의 시작일 때뿐입니다. 위의 세 경우는 실습 컨테이너에서 렌더링하거나 timetable로 계산해 확인했습니다.
4. Python 함수에는 값을 어떻게 전달할까?
책 4.2.3~4.2.4절 · 그림 4.5~4.10 · 리스트 4.5~4.13
PythonOperator 템플릿 (4.2.3)
BashOperator는 명령 문자열을 실행합니다. 반면 PythonOperator는 호출할 Python 함수를 받습니다.
def show_hour(data_interval_start):
print(data_interval_start.in_timezone("UTC").hour)
show = PythonOperator(
task_id="show_hour",
python_callable=show_hour,
)
python_callable=show_hour처럼 함수 자체를 넘깁니다. show_hour()로 호출한 결과를 넣는 방식이 아닙니다.
이 코드에는 {{ ... }}도 op_kwargs도 없습니다. 그런데 9~10시 실행에서 show_hour는 9를 출력합니다. 인자 이름을 보고 값을 채워 주는 것은 PythonOperator입니다. 태스크를 실행할 때 다음 순서로 함수를 부릅니다.
- 함수의 인자 이름, 즉 함수 괄호 안에 적은 이름을 읽습니다. 여기서는
data_interval_start입니다. - 실행 콘텍스트에서 같은 이름의 값을 찾습니다. 9~10시 실행이면 9시를 가리키는 날짜 객체입니다.
- 찾은 값을 넣어
show_hour(data_interval_start=<9시 날짜 객체>)처럼 부릅니다.
템플릿을 거치지 않고 날짜 객체가 그대로 들어오므로, 함수 안에서 .hour 같은 날짜 기능을 바로 쓸 수 있습니다.
- 이름은 정확히 같아야 합니다. 인자 이름을
data_interval_strat처럼 잘못 쓰면 콘텍스트에서 찾지 못해, 실습 환경에서TypeError: ... missing 1 required positional argument: 'data_interval_strat'로 실패했습니다. - Python 자체의 기능은 아닙니다. 같은 함수를 테스트 코드에서
show_hour()처럼 직접 부르면 아무 값도 채워지지 않습니다. 이때는show_hour(pendulum.datetime(2026, 10, 24, 9, tz="UTC"))처럼 날짜를 직접 넘깁니다.
필요한 이름을 명시하는 대신, 다음처럼 여러 키워드 인자를 받을 수도 있습니다.
def show_context(**context):
print(context["data_interval_start"])
이렇게 받으면 PythonOperator가 넘기는 값이 전부 context라는 사전 하나에 모입니다. 3절의 실행 콘텍스트를 통째로 받는 방법입니다. context는 작성자가 붙인 이름입니다. 앞의 별표 두 개는 Python이 여러 키워드 인자를 사전으로 모으는 문법이며, context라는 변수명 자체에 특별한 기능이 있는 것은 아닙니다.
provide_context와 공개 SDK책은
provide_context=True를 Airflow 1 방식으로 소개하면서, Airflow 2는 호환을 위해 이 인수를 받아 준다고 설명합니다. Airflow 3.3.2는 받아 주지 않습니다. 넘기면TypeError: Invalid arguments were passed to PythonOperator가 나서 DAG를 불러오지 못합니다. 필요한 콘텍스트 이름을 함수 인자로 받으면 됩니다.태스크 함수 안에서 현재 콘텍스트를 직접 가져올 수도 있습니다. Airflow 3의 공개 경로는
from airflow.sdk import get_current_context입니다. 태스크가 실행 중일 때 부르는 함수이며, DAG 파일의 최상위 코드에서 부르는 함수가 아닙니다.
PythonOperator에 변수 제공 (4.2.4)
출력 경로처럼 작성자가 정하는 값은 op_args나 op_kwargs로 전달합니다. 둘의 차이는 값을 함수 인자에 짝짓는 기준입니다. 다음 함수에 두 값을 넘긴다고 해 봅시다.
def download(url, output_path):
...
op_args는 순서로 짝짓습니다.op_args=["https://…/a.gz", "/tmp/a.gz"]라면 첫 번째 값이 첫 번째 인자url에, 두 번째 값이output_path에 들어갑니다.download("https://…/a.gz", "/tmp/a.gz")로 부르는 것과 같습니다. 리스트의 순서를 바꾸면 값도 다른 인자로 들어갑니다.op_kwargs는 이름으로 짝짓습니다.op_kwargs={"output_path": "/tmp/a.gz", "url": "https://…/a.gz"}라면 키 이름과 같은 인자에 값이 들어갑니다.download(url="https://…/a.gz", output_path="/tmp/a.gz")로 부르는 것과 같고, 사전에 적은 순서는 상관없습니다.
값이 두세 개를 넘으면 어느 값이 어느 인자로 가는지 보이는 op_kwargs가 읽기 쉽습니다. 이번 실습도 op_kwargs를 씁니다.
op_kwargs와 콘텍스트는 둘 다 이름으로 함수 인자를 채웁니다. PythonOperator는 함수를 부르기 전에 두 값을 사전 하나로 합친 뒤, 함수 인자 이름과 같은 값만 골라 넘깁니다. 마지막 실습의 _prepare_sample로 보면 다음과 같습니다.
콘텍스트 ds, data_interval_start, data_interval_end, ...
op_kwargs output_path
합친 사전 ds, data_interval_start, data_interval_end, ..., output_path
함수 인자 _prepare_sample(data_interval_end, output_path)
넘기는 값 data_interval_end, output_path
앞의 show_context처럼 값을 사전 하나로 받는 함수에는 합친 사전 전체가 들어오므로 op_kwargs의 값도 함께 들어 있습니다. 합칠 때 같은 이름이 있으면 op_kwargs 쪽 값이 남습니다. 그래서 op_kwargs에 콘텍스트와 같은 이름을 쓰면 콘텍스트의 값이 가려집니다. 실습 환경에서 op_kwargs={"ds": "my-own-value"}를 주자, 함수의 ds 인자에는 실행 날짜 대신 'my-own-value'가 들어왔습니다. 사용자 인자는 output_path·pages처럼 콘텍스트와 겹치지 않는 이름으로 짓습니다.
op_args·op_kwargs에 Jinja를 넣으면 문자열을 먼저 렌더링한 뒤 함수를 호출합니다.
def show_partition(partition):
print(partition)
show = PythonOperator(
task_id="show_partition",
python_callable=show_partition,
op_kwargs={
"partition": "{{ data_interval_start.in_timezone('UTC').strftime('%Y%m%d-%H') }}",
},
)
이 함수가 받는 partition은 '20261024-09'라는 글자입니다. 앞의 show_hour와 나란히 놓고 보면 차이가 어디서 생기는지 보입니다.
# ① 콘텍스트 이름을 함수 인자로 씀 → 날짜 객체
def show_hour(data_interval_start):
print(data_interval_start.hour) # 9
# ①' 콘텍스트를 사전으로 받아 꺼냄 → 날짜 객체
def show_hour_from_context(**context):
print(context["data_interval_start"].hour) # 9
# ② op_kwargs에 템플릿을 씀 → 글자
def show_partition(partition):
print(partition) # '20261024-09'
op_kwargs={"partition": "{{ data_interval_start.strftime('%Y%m%d-%H') }}"}
갈리는 곳은 값이 {{ ... }}를 거쳤는가입니다.
- ①·①’은 템플릿을 거치지 않습니다.
PythonOperator가 콘텍스트의 값을 그대로 넘기므로 날짜 객체가 들어옵니다. 인자 이름으로 받든(①), 사전으로 받아context["data_interval_start"]로 꺼내든(①’) 같습니다. - ②는 템플릿을 거칩니다.
op_kwargs는 템플릿 필드라서, 태스크를 실행하기 직전에 Airflow가{{ ... }}를 렌더링합니다. 중괄호 안에서는data_interval_start가 날짜 객체라.strftime()을 쓸 수 있지만, 렌더링의 결과는'20261024-09'라는 글자입니다. 함수에는 이 결과가 들어옵니다.
시계를 직접 건네면 받는 쪽도 시계를 받습니다. 누군가 시계를 보고 시각을 종이에 적어 건네면 받는 쪽은 종이를 받습니다. ①·①’은 시계(날짜 객체)를, ②는 종이(글자)를 받는 경우이고, 종이에 적는 일이 렌더링입니다.
둘 다 "9~10시 구간"이라는 같은 정보를 담고 있지만 할 수 있는 일이 다릅니다.
- 날짜 객체(
show_hour의data_interval_start):.hour로 시각을 꺼내고,strftime()으로 형식을 바꾸고, 시간대를 바꿀 수 있습니다. - 렌더링된 글자(
show_partition의partition): 이미 완성된 문자열이라 경로에 그대로 붙이는 데 씁니다.partition.hour를 부르면AttributeError: 'str' object has no attribute 'hour'가 납니다.
글자로 받는 것이 문제는 아닙니다. 경로처럼 글자로 쓸 값이면 글자로 받으면 되고, 글자에 .hour처럼 날짜 기능을 부를 때만 오류가 납니다. 그래서 받는 방법은 값으로 무엇을 할지 보고 고릅니다.
- 시각을 꺼내거나 계산할 값: 콘텍스트 이름을 인자로 써서 날짜 객체로 받습니다.
- 경로처럼 완성된 글자로 쓸 값:
op_kwargs의 템플릿으로 받습니다. 경로를 함수 안에서 날짜 객체로 만들어도 동작은 같지만, 템플릿으로 넘기면 5절의airflow tasks render로 실행 전에 경로를 확인할 수 있습니다. 실행한 뒤에는 웹 UI에서도 태스크가 시작할 때 채워진 경로를 볼 수 있습니다.
그림의 두 길은 둘 다 실행 콘텍스트에서 출발합니다. 위쪽은 템플릿을 거쳐 글자가 되어 들어가는 길이고, 아래쪽은 날짜 객체가 그대로 들어가는 길입니다.
한 함수에서 두 길을 함께 쓸 수도 있습니다. 마지막 실습의 _prepare_sample(data_interval_end, output_path)가 그렇습니다.
data_interval_end는 콘텍스트에 있는 이름이라 날짜 객체로 들어옵니다. 함수는 여기서.in_timezone("UTC")로 시각을 꺼내 파일 주소를 만듭니다.output_path는op_kwargs에 적은 이름이라 렌더링된 글자로 들어옵니다. 함수는 이 글자를 파일 경로로만 씁니다.
렌더링된 숫자도 문자열입니다
op_kwargs={"hour": "{{ data_interval_start.hour }}"}처럼 숫자를 렌더링해도, 기본 설정에서 함수는 숫자 9가 아니라 글자 '9'를 받습니다. 함수에서 hour + 1을 계산하면 TypeError: can only concatenate str (not "int") to str가 납니다. 계산할 값은 날짜 객체로 받습니다.
DAG에
render_template_as_native_obj=True를 설정하면, 렌더링된 글자가 Python 값처럼 생겼을 때 그 값으로 바꿔서 넘깁니다. 다만 모양만 보고 바꿉니다. 실습 환경에서{{ data_interval_start.strftime('%H') }}는 23시 실행에서 숫자23이 됐지만, 9시 실행에서는 글자'09'로 남았습니다. 앞에 0이 붙은09는 Python에서 숫자로 쓸 수 없는 모양이기 때문입니다. 같은 태스크가 시간에 따라 다른 타입을 받게 되므로, 이번 실습은 기본 설정(글자)을 유지합니다.
5. URL과 경로를 실행 전에 확인하기
책 4.2.5절 · 리스트 4.14 · 그림 4.11
URL에서 월·일의 0이 빠졌거나, 경로에 {{ ... }}가 그대로 남아 있다면 실제 다운로드를 오래 기다리기 전에 확인하고 싶습니다. 이때 렌더링된 템플릿을 봅니다.
마지막 실습 DAG가 등록되어 있다면 다음 명령으로 확인할 수 있습니다.
docker compose exec airflow airflow tasks render \
hourly_pageviews summarize '2026-10-24T10:00:00+00:00'
실습 환경에서 op_kwargs는 다음 값으로 렌더링됐습니다.
{
"input_path": "/tmp/ch04/20261024-09/pageviews.gz",
"output_path": "/tmp/ch04/20261024-09/summary.json",
"pages": ["Apple", "Google", "Amazon"],
"interval_start": "2026-10-24T09:00:00+00:00",
"interval_end": "2026-10-24T10:00:00+00:00",
}
이 명령은 태스크의 템플릿을 렌더링해 보여 줍니다. 집계 함수를 실행하거나, 경로에 파일이 실제로 있는지 확인하는 명령은 아닙니다. 값이 맞는지를 먼저 보고, 파일 생성과 집계는 DAG를 실행해 확인합니다.
Airflow 3.3.2의
tasks render는 논리적 날짜나 실행 ID(run_id)를 받습니다. 위 명령의10:00은 직접 실행의 논리적 날짜라서, 데이터 구간은 3절에서 본 것처럼 그 직전의 09~10시가 됩니다. 명령 끝의 시각을 구간의 시작으로 읽지 말고, 출력의data_interval_start·data_interval_end를 확인합니다.
태스크가 실행된 뒤에는 웹 UI에서 그 실행에 채워진 값을 볼 수 있습니다. 소스의 원래 문자열과 실행에 채워진 값을 나눠 보면, 날짜 선택이 틀렸는지 처리 코드가 틀렸는지 좁히기 쉽습니다.
6. 외부 데이터베이스에는 어떻게 연결할까?
책 4.3절 · 그림 4.12~4.15 · 리스트 4.15~4.22 · 표 4.2
템플릿으로 시간별 파일을 골랐다면, 이제 필요한 페이지의 조회 수를 저장합니다. 원본은 압축을 풀고 관심 페이지의 영어 위키피디아 행만 골라, PostgreSQL에서 실행할 INSERT 문을 파일로 만듭니다. 그다음 PostgresOperator가 그 SQL 파일을 읽어 실행합니다.
여기서 PostgreSQL은 분석 결과를 저장하는 외부 DB입니다. Airflow가 태스크 상태와 실행 기록을 보관하는 메타데이터 DB와 역할이 다릅니다. 이 실습 저장소의 메타데이터 DB는 SQLite입니다. PostgreSQL에 결과를 쓰려면 PostgreSQL 서버를 따로 준비하고, 아래의 Connection으로 접속 정보를 알려 줘야 합니다.
Connection·Hook·Operator의 역할
서버 주소·포트·DB 이름·사용자·비밀번호를 DAG마다 직접 쓰는 대신, Connection에 접속 정보를 등록하고 코드에서는 연결 ID를 참조합니다.
연결 ID는 이 접속 정보 묶음에 붙인 이름입니다. 코드에는 conn_id="ch04_pageviews"처럼 이름만 적고, 주소와 비밀번호는 Airflow가 실행할 때 찾아 씁니다. 어떤 테이블에 접근할 수 있는지는 연결 ID가 아니라, 그 안에 적은 DB 계정의 권한이 정합니다.
연결 정보는 UI·CLI로 등록하거나 환경변수로 줄 수 있습니다. 연결 ID가 ch04_pageviews라면 환경변수 이름은 AIRFLOW_CONN_CH04_PAGEVIEWS입니다. 등록 방식은 공식 Connection 관리 문서에 있습니다.
| 구성 요소 | 담당하는 일 | 예시 |
|---|---|---|
| Connection | 어디에 어떤 계정으로 접속할지 보관 | 연결 ID ch04_pageviews, 호스트, DB 이름 |
| Hook | 연결 정보를 사용해 외부 시스템과 통신 | PostgreSQL 연결·쿼리 실행 |
| Operator | Airflow 태스크로 수행할 작업을 정의 | 지정한 SQL 실행, 성공·실패 관리 |
여러 DAG가 같은 연결 ID를 쓰면, 비밀번호나 주소가 바뀌어도 Connection 하나만 고치면 됩니다.
PostgresOperator대신 공통 SQL 오퍼레이터이번 컨테이너의 PostgreSQL 프로바이더는 7.0.2이며, 책의
PostgresOperator경로는 더 이상 제공되지 않습니다. SQL 실행에는airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator를 사용하고postgres_conn_id대신conn_id를 지정합니다. PostgreSQL 연결을 처리하는 프로바이더도 필요합니다.
예를 들어 ch04_pageviews가 PostgreSQL 연결로 등록되어 있다면 다음 태스크로 연결과 값 전달을 확인할 수 있습니다.
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
query = SQLExecuteQueryOperator(
task_id="query",
conn_id="ch04_pageviews",
sql="SELECT %(page)s AS page, %(views)s::integer AS views, '{{ ds }}' AS day;",
parameters={"page": "Apple", "views": 15},
)
두 가지 치환이 함께 보입니다. {{ ds }}는 Airflow의 Jinja가 실행 콘텍스트로 렌더링합니다. %(page)s·%(views)s는 PostgreSQL 드라이버에 넘긴 SQL 매개변수입니다. 페이지 제목 같은 데이터 값은 SQL 문자열에 직접 이어 붙이기보다 parameters로 전달합니다.
이 예제는 별도 PostgreSQL 16 컨테이너에 연결해 ('Apple', 15, '2026-10-24')가 반환되는 것까지 확인했습니다. 전체 위키미디어 덤프를 적재한 검증은 아니며, 아래 기본 실습은 외부 DB 없이 돌아가게 만들었습니다.
SQL 파일도 템플릿이 될 수 있습니다
책의 PostgresOperator는 sql="postgres_query.sql"처럼 SQL 문장 대신 파일 이름을 받습니다(리스트 4.20). 값이 .sql로 끝나면 Airflow는 이것을 파일 이름으로 보고, 파일을 읽어 그 안의 {{ ... }}를 렌더링한 뒤 SQL로 실행합니다. 어떤 확장자를 파일로 볼지는 오퍼레이터의 template_ext가 정하며, 이번 컨테이너의 SQLExecuteQueryOperator는 .sql·.json입니다.
Airflow가 이 파일을 찾는 폴더는 정해져 있습니다. 기본으로는 DAG 파일이 있는 폴더에서 찾고, 다른 폴더도 찾게 하려면 DAG의 template_searchpath에 적습니다. 책은 앞 태스크가 INSERT 문을 /tmp/postgres_query.sql로 만들기 때문에 template_searchpath="/tmp"를 지정합니다. /tmp가 특별해서가 아니라, 앞 태스크가 파일을 그곳에 썼기 때문입니다.
이 방식은 앞 태스크가 쓴 파일을 뒤 태스크가 읽으므로, 두 태스크가 같은 디스크를 봐야 합니다. 이 저장소는 컨테이너 하나에서 모든 태스크가 돌아 문제가 없습니다. 하지만 태스크가 서로 다른 워커(서버)에서 돌면, 워커 A의 /tmp에 쓴 파일이 워커 B의 /tmp에는 없습니다. 이럴 때는 공유 스토리지처럼 모든 워커가 보는 곳에 파일을 둡니다. 고정된 쿼리라면 SQL 파일을 DAG와 같은 폴더에 두고 sql="sql/pageviews.sql"처럼 참조하면 됩니다.
결과 테이블에는 페이지 제목·조회 수와 함께 그 시간이 언제인지까지 저장합니다. ds만 저장하면 같은 날의 여러 시간을 구분할 수 없습니다. 같은 시간을 다시 처리할 때 행이 중복되지 않도록, 시간·페이지를 기준으로 교체하는 방식도 정해야 합니다.
7. 직접 돌려 보기: 시간별 샘플 페이지뷰 집계
책 4.2~4.3절 예제를 Airflow 3에서 확인하는 실습
먼저 저자 저장소의 리스트 4.13에는 다음 템플릿이 들어 있습니다.
# 책 원본 코드 일부 (Airflow 2.0 기준, op_kwargs 중 두 항목)
op_kwargs = {
"year": "{{ execution_date.year }}",
"hour": "{{ execution_date.hour }}",
}
원본의 아이디어는 시간별 파일 주소를 실행 날짜로 채우는 것입니다. Airflow 3에서는 다음 부분을 바꿉니다.
| 원본에서 확인할 부분 | 이번 실습에서 바꾸는 방법 |
|---|---|
from airflow import DAG |
공개 SDK의 from airflow.sdk import DAG |
| 이전 오퍼레이터 import 경로 | airflow.providers.standard.operators... |
days_ago(1) |
시간대가 있는 고정 시작 날짜 |
schedule_interval="@hourly" |
schedule=CronDataIntervalTimetable("0 * * * *", timezone="UTC") |
execution_date 템플릿 |
파일 주소는 data_interval_end, 저장 경로는 data_interval_start |
| 모든 실행이 같은 임시 파일 경로 | UTC 날짜·시간별 디렉터리 사용 |
| 큰 외부 덤프 다운로드·PostgreSQL 적재 | 작은 로컬 압축 샘플 생성·JSON 집계로 동작 확인 |
schedule="@hourly"로만 바꾸면 Airflow 3 기본 설정에서는 시작과 끝이 같은, 길이 0인 구간이 만들어집니다(3장). 한 시간짜리 구간이 필요하므로 timetable을 직접 지정했습니다. 마지막 두 줄은 Airflow 3 호환성 때문에 필수인 변경은 아닙니다. 실행마다 결과가 섞이지 않게 하고, 큰 파일 다운로드나 별도 DB 없이 템플릿 동작을 확인하기 위한 선택입니다.
호환성 검사부터 실행합니다.
uvx ruff@latest check --select AIR3 dags/
이 검사는 이전 import 경로 같은 문제를 찾는 데 유용하지만, 파일 주소가 맞는지나 SQL이 실제 DB에서 실행되는지까지 보장하지는 않습니다. 이어서 실제 DAG를 실행합니다.
Airflow 3용 코드
아래 코드를 dags/ch04_hourly_pageviews.py로 저장합니다. 외부 다운로드는 하지 않고, 날짜에 맞는 URL만 로그에 표시합니다. 그다음 형식이 같은 작은 .gz 샘플을 만들어 집계합니다.
"""Chapter 4: render hourly pageview paths with Airflow 3.
Inspired by (not copied):
https://github.com/BasPH/data-pipelines-with-apache-airflow/tree/master/chapter04
Uses a tiny local fixture instead of downloading the full Wikimedia dump.
All tasks share /tmp inside this repository's single Airflow container.
The Wikimedia file name marks the END of its UTC hour, so the URL uses
data_interval_end while the local partition uses data_interval_start.
"""
import gzip
import json
from pathlib import Path
import pendulum
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG, CronDataIntervalTimetable
# Wikimedia 파일은 UTC 시각 기준입니다. 같은 시간은 같은 경로를 사용합니다.
PARTITION = "{{ data_interval_start.in_timezone('UTC').strftime('%Y%m%d-%H') }}"
ROOT = "/tmp/ch04/" + PARTITION
SAMPLE_PATH = ROOT + "/pageviews.gz"
SUMMARY_PATH = ROOT + "/summary.json"
def _prepare_sample(data_interval_end, output_path):
# 이 인자는 Jinja 문자열이 아니라 PythonOperator가 전달한 날짜 객체입니다.
# 위키미디어 파일 이름의 시각은 집계 구간의 끝이라 구간 끝으로 주소를 만듭니다.
hour = data_interval_end.in_timezone("UTC")
url = (
"https://dumps.wikimedia.org/other/pageviews/"
f"{hour:%Y}/{hour:%Y-%m}/pageviews-{hour:%Y%m%d-%H}0000.gz"
)
print(f"source_url={url} (sample only; no download)")
print(f"context_type={type(data_interval_end).__name__}")
# 함수 본문의 일반 문자열까지 Airflow가 찾아서 치환하지는 않습니다.
print("ordinary_python_string={{ ds }}")
# 실제 파일처럼 프로젝트·제목 순으로 정렬하고, 같은 쌍은 한 번만 둡니다.
rows = (
"en Amazon 8 0\n"
"en Apple 15 0\n"
"en Facebook 2 0\n"
"en Google 7 0\n"
"en Microsoft 4 0\n"
"en.m Apple 5 0\n"
"nl Apple 99 0\n"
)
path = Path(output_path)
path.parent.mkdir(parents=True, exist_ok=True)
with gzip.open(path, "wt", encoding="utf-8") as stream:
stream.write(rows)
print(f"wrote {path}")
def _summarize(input_path, output_path, pages, interval_start, interval_end):
counts = dict.fromkeys(pages, 0)
with gzip.open(input_path, "rt", encoding="utf-8") as stream:
for line in stream:
project, title, views, _ = line.split()
if project == "en" and title in counts:
counts[title] = int(views)
result = {
"interval_start": interval_start,
"interval_end": interval_end,
"pageviews": counts,
}
text = json.dumps(result, indent=2, sort_keys=True)
Path(output_path).write_text(text + "\n", encoding="utf-8")
print(text)
print(f"wrote {output_path}")
with DAG(
dag_id="hourly_pageviews",
start_date=pendulum.datetime(2026, 10, 24, tz="UTC"),
schedule=CronDataIntervalTimetable("0 * * * *", timezone="UTC"),
catchup=False,
max_active_runs=1,
tags=["book", "chapter04"],
):
prepare_sample = PythonOperator(
task_id="prepare_sample",
python_callable=_prepare_sample,
op_kwargs={"output_path": SAMPLE_PATH},
do_xcom_push=False,
)
inspect_template = BashOperator(
task_id="inspect_template",
# 날짜는 env에서 렌더링하고 셸에서는 변수를 따옴표로 감싸 사용합니다.
bash_command='printf "partition=%s\nsample=%s\n" "$PARTITION" "$SAMPLE_PATH"; test -s "$SAMPLE_PATH"',
env={"PARTITION": PARTITION, "SAMPLE_PATH": SAMPLE_PATH},
append_env=True,
do_xcom_push=False,
)
summarize = PythonOperator(
task_id="summarize",
python_callable=_summarize,
op_kwargs={
"input_path": SAMPLE_PATH,
"output_path": SUMMARY_PATH,
"pages": ["Apple", "Google", "Amazon"],
"interval_start": "{{ data_interval_start.isoformat() }}",
"interval_end": "{{ data_interval_end.isoformat() }}",
},
do_xcom_push=False,
)
prepare_sample >> inspect_template >> summarize
코드를 읽는 순서
- 경로의 날짜를 정합니다.
PARTITION은 UTC 구간 시작으로20261024-09같은 값을 만듭니다. 같은 구간을 다시 실행하면 같은 경로를 사용합니다. - 객체와 문자열을 나눠 받습니다. 4절에서 본 대로
_prepare_sample은 날짜 객체(data_interval_end)로 파일 주소를 만들고, 렌더링된 글자(output_path)를 경로로 씁니다. - Bash에서도 같은 경로를 확인합니다.
inspect_template은 템플릿 필드인env에 날짜와 경로를 채워 셸에 전달합니다. 셸은 따옴표로 감싼 환경변수를 출력하고 파일이 있는지 확인합니다. - 필요한 행만 고릅니다.
_summarize는en프로젝트의 Apple·Google·Amazon 행만 골라 조회 수를 기록합니다. 모바일en.m Apple과 네덜란드어nl Apple은 제외됩니다. - 작업 순서와 파일 연결을 구분합니다.
>>는 실행 순서를 정합니다. 파일은 세 태스크에 같은 경로를 전달해 연결하며,>>가 파일 내용을 옮겨 주는 것은 아닙니다.
이 실습은 콘텍스트와 템플릿에 집중한 작은 예제입니다. 샘플 파일과 JSON을 직접 쓰므로, 3장의 임시 파일 후 교체 방식까지 적용한 저장 구현은 아닙니다.
실행하기
실습 환경은 airflow-book-study 저장소의 Docker 구성을 사용합니다. 아직 4장 파일이 없다면 위 코드를 먼저 저장합니다.
docker compose up -d
이번 DAG는 매시간의 데이터 구간을 만들고 catchup=False로 최신 완료 구간부터 처리합니다. 특정 시간의 결과를 비교하려면 다음처럼 한 번 테스트할 수 있습니다.
docker compose exec airflow airflow dags test \
hourly_pageviews '2026-10-24T10:00:00+00:00'
이 수동 테스트에서 확인한 데이터 구간은 UTC 09:00 이상 ~ 10:00 미만입니다. 결과 파일도 09시 디렉터리에 생깁니다.
docker compose exec airflow cat /tmp/ch04/20261024-09/summary.json
{
"interval_end": "2026-10-24T10:00:00+00:00",
"interval_start": "2026-10-24T09:00:00+00:00",
"pageviews": {
"Amazon": 8,
"Apple": 15,
"Google": 7
}
}
앞의 tasks render 명령으로 이 경로를 렌더링해 보는 것과, dags test로 파일을 실제로 만드는 것을 나눠 확인할 수 있습니다.
실제로 돌린 결과
Airflow 3.3.2 공식 Docker 이미지에서 확인했습니다(샘플 DAG는 2026년 11월 1일, SQL 연결 검사는 10월 30일). standard 프로바이더는 1.19.0, common-sql은 2.1.1, PostgreSQL 프로바이더는 7.0.2였습니다.
- 샘플 DAG:
prepare_sample → inspect_template → summarize세 태스크가 모두 성공했습니다. 샘플 압축 파일과 JSON이 시간별 디렉터리에 생겼습니다. - 콘텍스트 객체: 함수 인자로 받은 구간 끝은
DateTime이었고, 이 값으로pageviews-20261024-100000.gz주소를 만들었습니다. 실제 다운로드 요청은 보내지 않았습니다. - 렌더링 범위:
env와op_kwargs의 템플릿은 날짜·경로로 바뀌었습니다. 함수 본문의 일반 문자열ordinary_python_string={{ ds }}는 그대로 로그에 남았습니다. - 집계 결과: Apple 15, Amazon 8, Google 7이었습니다. 모바일(
en.m) Apple 5건, 네덜란드어(nl) Apple 99건, 고르지 않은 Microsoft·Facebook은 결과에 들어가지 않았습니다. - 별도 SQL 연결 검사: 로컬 PostgreSQL 16과 연결 정보를 사용해 조회 태스크를 실행했습니다. Jinja 날짜 렌더링과 SQL 매개변수 전달, 외부
.sql파일의 템플릿 렌더링을 확인했습니다.
8. 4장을 마치며
처음의 다섯 질문에 답해 보면 이렇습니다.
- 실행한 시각과 맡은 시간은 다릅니다. 9~10시를 맡은 실행은 10시가 지나서 시작하므로
datetime.now()는 10시 이후를 가리키고, 늦게 돌거나 다시 돌리면 더 멀어집니다.data_interval_start(09:00)와data_interval_end(10:00)는 실행이 만들어질 때 정해져 바뀌지 않습니다.logical_date는 예약 실행에서는 09:00이지만 날짜를 지정해 직접 실행하면 그 시각이 되므로, 처리할 시간은 데이터 구간에서 꺼냅니다. - 구간의 끝은 직접 더하지 않고
data_interval_end에서 꺼냅니다. Airflow가 실행을 만들 때 스케줄을 보고 계산해 둔 값이라, 직접 실행하거나 평일에만 도는 DAG처럼logical_date에 한 시간·하루를 더하면 틀리는 경우에도 맞습니다. 이번 예제는 위키미디어 파일 이름이 구간의 끝 시각이라 파일 주소에data_interval_end를, 결과 폴더에data_interval_start를 썼습니다. - 템플릿은 각 태스크를 실행하기 직전에 렌더링합니다. 오퍼레이터의
template_fields에 포함된 속성만 대상이며, Python 함수 본문의 일반 문자열까지 바꾸지는 않습니다.airflow tasks render로 채워진 값을 미리 볼 수 있지만, 경로가 렌더링됐다고 파일이 만들어진 것은 아니므로 실행 결과와 나눠 확인합니다. - 인자 이름을 보고 콘텍스트 값을 채워 주는 것은
PythonOperator입니다. 인자 이름이data_interval_start면 날짜 객체를 받고,op_kwargs의 템플릿으로 넘기면 렌더링된 글자를 받습니다.op_args는 순서로,op_kwargs는 이름으로 함수 인자에 짝지어집니다. - Connection은 접속 정보, Hook은 외부 시스템 통신, Operator는 태스크의 작업을 맡습니다. 현재 프로바이더에서 지원하는 오퍼레이터와 연결 ID를 사용합니다.
3장에서 “이번 실행이 어느 시간의 데이터를 맡는가”를 정했다면, 4장에서는 그 시간을 URL·파일 경로·함수 인자로 전달해 실제 처리 코드와 연결했습니다. 같은 DAG 정의를 여러 시간에 사용할 수 있는 이유가 여기에 있습니다.
다음 5장(태스크 간 의존성 정의하기)에서는 일렬로 이어진 순서를 넘어, 여러 태스크를 모으고 나누는 의존성, 조건에 따른 분기, 트리거 규칙, XCom과 TaskFlow API를 살펴봅니다.
이 글은 Apache Airflow 한국 사용자 모임 운영진으로 활동하며 《Apache Airflow 기반의 데이터 파이프라인》을 학습하고, Airflow 3에서 달라진 내용을 더해 정리한 기록입니다. 글을 읽으시며 의견이나 함께 나누고 싶은 인사이트가 있다면 언제든 댓글로 남겨주세요!