# \[Airflow 책 4장\] Airflow 콘텍스트를 사용하여 태스크 템플릿 작업하기: datetime.now()로 날짜를 넣으면 안 될까? (Airflow 3 반영)

**URL:** <https://discourse.airflow-kr.org/t/airflow-4-airflow-datetime-now-airflow-3/657>\
**Category:** 정보공유\
**Tags:** study\
**Created:** [11월 12, 2026, 5:47오전 UTC](https://discourse.airflow-kr.org/t/airflow-4-airflow-datetime-now-airflow-3/657 "2026-11-12T05:47:26Z")\
**Posts on this page:** 1\
**Page:** 1

<div class="post-metadata">

**Author:** ![vividbaek](https://discourse.airflow-kr.org/user_avatar/discourse.airflow-kr.org/vividbaek/32/811_2.png) [@vividbaek](https://discourse.airflow-kr.org/u/vividbaek)\
**Post date:** [11월 12, 2026, 5:47오전 UTC](https://discourse.airflow-kr.org/t/airflow-4-airflow-datetime-now-airflow-3/657/1 "2026-11-12T05:47:27Z")

</div>

3장 실습에서는 파일 경로에 `{{ data_interval_start }}`를 넣었습니다. 그런데 그냥 `datetime.now()`로 현재 시각을 넣으면 안 될까요? 매시간 도는 DAG에서 10시가 조금 지나 시작한 실행은 10시가 아니라 9~10시의 데이터를 처리합니다. 그렇다면 이 "9시"는 어디서 받고, `{{ ... }}` 문자열과 Python 함수에는 언제 어떻게 채워질까요?

《Apache Airflow 기반의 데이터 파이프라인》 4장은 위키피디아의 시간별 페이지뷰를 가져오는 예제로 이 과정을 설명합니다. 날짜를 URL에 넣고, 필요한 페이지의 조회 수를 고른 뒤, 결과를 외부 데이터베이스에 저장합니다.

**📌 이 글에서 알아갈 수 있는 것**

1. 9~10시 데이터를 처리하는 실행에서 `datetime.now()`·`logical_date`·`data_interval_start`는 **각각 몇 시일까?**
2. 구간의 끝이 필요하면 **`logical_date`에 한 시간을 더하면 될까?**
3. `{{ ... }}`는 **언제 실제 값으로 바뀌고** , 실행 전에 어떻게 확인할까?
4. `PythonOperator`의 함수는 날짜를 **객체로 받을까, 문자열로 받을까?** `op_args`와 `op_kwargs`는 어떻게 다를까?
5. 외부 DB에 접속할 때 **Operator·Hook·Connection은 각각 무엇을 할까?**

이번 장에서 연결할 핵심은 **DAG를 정의할 때 작업의 틀을 만들고, 각 태스크를 실행할 때 그 실행에 맞는 값으로 틀을 채운다** 는 것입니다. 이때 값을 제공하는 묶음이 **태스크 콘텍스트** , 문자열의 빈칸을 채우는 도구가 **Jinja 템플릿** 입니다.

 ![image](https://discourse.airflow-kr.org/uploads/default/original/2X/2/22ec0b9191eb66a175a5f5e6577ddc8d0f9474ec.jpeg)

* * *

### 📊 1. 무엇을 만들까: 시간별 위키피디아 페이지뷰

_📖 책 4.1~4.1.1절 · 그림 4.1~4.2_

예제의 배경은 가상의 주식시장 예측 도구 StockSense입니다. 어떤 기업의 위키피디아 페이지뷰가 늘면 그 기업에 대한 관심이 커져 주가도 오를 수 있다는 가정에서 출발합니다. 이번 장은 이 가정을 검증하지 않고, 검증에 쓸 Apple·Amazon·Google 같은 기업 페이지의 조회 수를 모으는 파이프라인까지 만듭니다.

위키미디어는 페이지뷰를 시간별 압축 파일로 공개합니다. **UTC** 기준 2019년 7월 17일 17~18시의 조회 수를 담은 파일 주소는 다음과 같습니다.

```plaintext
https://dumps.wikimedia.org/other/pageviews/2019/2019-07/pageviews-20190717-180000.gz

```

파일 이름의 `180000`은 집계 구간의 **끝** 시각입니다(그림 4.1). 그래서 18시 파일에는 17시부터 18시 전까지의 조회 수가 들어 있습니다. 이 점은 3절에서 URL을 만들 때 다시 씁니다.

URL의 연·월·일·시각을 바꾸면 다른 시간의 파일을 고를 수 있습니다. 3장에서 API의 조회 시작·종료 조건을 바꿨다면, 이번에는 **이미 시간별로 나뉜 파일 중 하나를 선택** 합니다.

파일 안의 행은 다음 형태입니다.

```plaintext
en Apple 15 0

```

| 항목 | 예시 | 이번에 쓰는 방법 |
| --- | --- | --- |
| 프로젝트 | `en` | 영어 위키피디아 행만 고릅니다. |
| 페이지 제목 | `Apple` | 관심 있는 페이지인지 고릅니다. |
| 조회 수 | `15` | 해당 시간의 조회 수로 저장합니다. |
| 응답 크기 | `0` | 위키미디어가 페이지뷰 파일에서는 집계를 빼서 0으로 나옵니다. 쓰지 않습니다. |

실제 시간별 파일에는 많은 페이지가 프로젝트·제목 순으로 정렬되어 있고, 같은 프로젝트·제목 쌍은 한 파일에 한 번만 나옵니다. 그래서 파일을 내려받은 뒤 필요한 행만 고르면 됩니다. 형식 설명은 [위키미디어 페이지뷰 문서](https://wikitech.wikimedia.org/wiki/Data_Platform/Data_Lake/Traffic/Pageviews), 파일 목록은 [페이지뷰 덤프](https://dumps.wikimedia.org/other/pageviews/)에서 확인할 수 있습니다.

#### 증분 데이터를 적재하는 방법 결정하기 (4.1.1)

필요한 시간마다 파일 하나를 고르면 됩니다. 실행할 때마다 모든 과거 파일을 다시 읽을 이유는 없습니다.

```plaintext
시간별 .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`에 다음 명령을 넣을 수 있습니다.

```python
BashOperator(
    task_id="show_day",
    bash_command="echo {{ ds }}",
)

```

DAG 파일을 읽고 이 오퍼레이터를 만들 때는 명령에 `{{ ds }}`가 남아 있습니다. **해당 태스크를 실행하기 전에** , Airflow가 그 실행의 콘텍스트를 사용해 템플릿을 렌더링합니다. 렌더링은 템플릿을 최종 값으로 바꾸는 과정입니다.

논리적 날짜가 2026년 10월 24일인 실행이라면 위 명령은 다음처럼 됩니다.

```bash
echo 2026-10-24

```

그다음에 셸이 이 명령을 실행합니다. Jinja가 명령을 실행하는 것은 아닙니다.

 ![image](https://discourse.airflow-kr.org/uploads/default/original/2X/c/c132bd970549021c31629c909bf2728d8ed42855.jpeg)

이 순서 때문에 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 템플릿 참조 문서](https://airflow.apache.org/docs/apache-airflow/3.3.2/templates-ref.html)에 정리되어 있습니다.

#### 폴더는 구간의 시작, 파일 주소는 구간의 끝

콘텍스트의 날짜는 연·월·일·시각을 꺼내거나 형식을 바꿀 수 있는 객체입니다. 이번 예제는 같은 9~10시 구간에서 두 값을 씁니다.

| 쓰는 곳 | 쓰는 값 | 9~10시 구간의 결과 |
| --- | --- | --- |
| 결과를 저장할 폴더 | `data_interval_start` | `20261024-09` |
| 내려받을 위키미디어 파일 | `data_interval_end` | `pageviews-20261024-100000.gz` |

폴더는 "어느 시간의 결과인가"를 나타내므로 구간의 시작인 9시로 이름을 붙입니다.

```jinja
{{ data_interval_start.in_timezone('UTC').strftime('%Y%m%d-%H') }}

```

파일 주소에는 **구간의 끝** 을 씁니다. 1절에서 본 것처럼 위키미디어는 집계 구간이 끝나는 시각을 파일 이름에 붙이기 때문에, 9~10시 조회 수는 `100000` 파일에 들어 있습니다. 마침 Airflow도 9~10시 구간의 실행을 10시가 지나서 시작하고, 그 실행의 `data_interval_end`가 바로 10:00입니다. 그래서 구간의 끝을 그대로 파일 이름에 넣으면 됩니다.

```jinja
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 함수** 를 받습니다.

```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`입니다.** 태스크를 실행할 때 다음 순서로 함수를 부릅니다.

1. 함수의 인자 이름, 즉 함수 괄호 안에 적은 이름을 읽습니다. 여기서는 `data_interval_start`입니다.
2. 실행 콘텍스트에서 같은 이름의 값을 찾습니다. 9~10시 실행이면 9시를 가리키는 날짜 객체입니다.
3. 찾은 값을 넣어 `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"))`처럼 날짜를 직접 넘깁니다.

필요한 이름을 명시하는 대신, 다음처럼 여러 키워드 인자를 받을 수도 있습니다.

```python
def show_context(**context):
    print(context["data_interval_start"])

```

이렇게 받으면 `PythonOperator`가 넘기는 값이 전부 `context`라는 사전 하나에 모입니다. 3절의 실행 콘텍스트를 통째로 받는 방법입니다. `context`는 작성자가 붙인 이름입니다. 앞의 별표 두 개는 Python이 여러 키워드 인자를 사전으로 모으는 문법이며, `context`라는 변수명 자체에 특별한 기능이 있는 것은 아닙니다.

> **💡 [Airflow 2.0 이후 달라진 점] `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`로 전달합니다. 둘의 차이는 **값을 함수 인자에 짝짓는 기준** 입니다. 다음 함수에 두 값을 넘긴다고 해 봅시다.

```python
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`로 보면 다음과 같습니다.

```plaintext
콘텍스트 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를 넣으면 **문자열을 먼저 렌더링한 뒤** 함수를 호출합니다.

```python
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`와 나란히 놓고 보면 차이가 어디서 생기는지 보입니다.

```python
# ① 콘텍스트 이름을 함수 인자로 씀 → 날짜 객체
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에서도 태스크가 시작할 때 채워진 경로를 볼 수 있습니다.

 ![image](https://discourse.airflow-kr.org/uploads/default/original/2X/e/e496e8ab7701594c100a8fd2a4ef31aaa14d0ee1.jpeg)

그림의 두 길은 둘 다 실행 콘텍스트에서 출발합니다. 위쪽은 템플릿을 거쳐 **글자** 가 되어 들어가는 길이고, 아래쪽은 **날짜 객체** 가 그대로 들어가는 길입니다.

한 함수에서 두 길을 함께 쓸 수도 있습니다. 마지막 실습의 `_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`가 납니다. 계산할 값은 날짜 객체로 받습니다.

> **💡 [Airflow 2.0 이후 달라진 점] Python 타입을 유지하는 템플릿**
> 
> 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가 등록되어 있다면 다음 명령으로 확인할 수 있습니다.

```bash
docker compose exec airflow airflow tasks render \
  hourly_pageviews summarize '2026-10-24T10:00:00+00:00'

```

실습 환경에서 `op_kwargs`는 다음 값으로 렌더링됐습니다.

```python
{
    "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 2.0 이후 달라진 점] 날짜 인자와 실제 데이터 구간**
> 
> 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 관리 문서](https://airflow.apache.org/docs/apache-airflow/3.3.2/howto/connection.html)에 있습니다.

| 구성 요소 | 담당하는 일 | 예시 |
| --- | --- | --- |
| Connection | 어디에 어떤 계정으로 접속할지 보관 | 연결 ID `ch04_pageviews`, 호스트, DB 이름 |
| Hook | 연결 정보를 사용해 외부 시스템과 통신 | PostgreSQL 연결·쿼리 실행 |
| Operator | Airflow 태스크로 수행할 작업을 정의 | 지정한 SQL 실행, 성공·실패 관리 |

 ![image](https://discourse.airflow-kr.org/uploads/default/original/2X/2/28bb47291a812ffdfbf2f911f3e29c6e3ccd36a0.jpeg)

여러 DAG가 같은 연결 ID를 쓰면, 비밀번호나 주소가 바뀌어도 Connection 하나만 고치면 됩니다.

> **💡 [Airflow 2.0 이후 달라진 점] `PostgresOperator` 대신 공통 SQL 오퍼레이터**
> 
> 이번 컨테이너의 PostgreSQL 프로바이더는 7.0.2이며, 책의 `PostgresOperator` 경로는 더 이상 제공되지 않습니다. SQL 실행에는 `airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator`를 사용하고 `postgres_conn_id` 대신 `conn_id`를 지정합니다. PostgreSQL 연결을 처리하는 프로바이더도 필요합니다.

예를 들어 `ch04_pageviews`가 PostgreSQL 연결로 등록되어 있다면 다음 태스크로 연결과 값 전달을 확인할 수 있습니다.

```python
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](https://github.com/BasPH/data-pipelines-with-apache-airflow/blob/master/chapter04/dags/listing_4_13.py)에는 다음 템플릿이 들어 있습니다.

```python
# 책 원본 코드 일부 (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 없이 템플릿 동작을 확인하기 위한 선택입니다.

호환성 검사부터 실행합니다.

```bash
uvx ruff@latest check --select AIR3 dags/

```

이 검사는 이전 import 경로 같은 문제를 찾는 데 유용하지만, 파일 주소가 맞는지나 SQL이 실제 DB에서 실행되는지까지 보장하지는 않습니다. 이어서 실제 DAG를 실행합니다.

#### Airflow 3용 코드

아래 코드를 `dags/ch04_hourly_pageviews.py`로 저장합니다. 외부 다운로드는 하지 않고, 날짜에 맞는 URL만 로그에 표시합니다. 그다음 형식이 같은 작은 `.gz` 샘플을 만들어 집계합니다.

```python
"""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

```

#### 코드를 읽는 순서

1. **경로의 날짜를 정합니다.** `PARTITION`은 UTC 구간 시작으로 `20261024-09` 같은 값을 만듭니다. 같은 구간을 다시 실행하면 같은 경로를 사용합니다.
2. **객체와 문자열을 나눠 받습니다.** 4절에서 본 대로 `_prepare_sample`은 날짜 객체(`data_interval_end`)로 파일 주소를 만들고, 렌더링된 글자(`output_path`)를 경로로 씁니다.
3. **Bash에서도 같은 경로를 확인합니다.** `inspect_template`은 템플릿 필드인 `env`에 날짜와 경로를 채워 셸에 전달합니다. 셸은 따옴표로 감싼 환경변수를 출력하고 파일이 있는지 확인합니다.
4. **필요한 행만 고릅니다.** `_summarize`는 `en` 프로젝트의 Apple·Google·Amazon 행만 골라 조회 수를 기록합니다. 모바일 `en.m Apple`과 네덜란드어 `nl Apple`은 제외됩니다.
5. **작업 순서와 파일 연결을 구분합니다.** `>>`는 실행 순서를 정합니다. 파일은 세 태스크에 같은 경로를 전달해 연결하며, `>>`가 파일 내용을 옮겨 주는 것은 아닙니다.

이 실습은 콘텍스트와 템플릿에 집중한 작은 예제입니다. 샘플 파일과 JSON을 직접 쓰므로, 3장의 임시 파일 후 교체 방식까지 적용한 저장 구현은 아닙니다.

#### 실행하기

실습 환경은 [airflow-book-study 저장소](https://github.com/vividbaek/airflow-book-study)의 Docker 구성을 사용합니다. 아직 4장 파일이 없다면 위 코드를 먼저 저장합니다.

```bash
docker compose up -d

```

이번 DAG는 매시간의 데이터 구간을 만들고 `catchup=False`로 최신 완료 구간부터 처리합니다. 특정 시간의 결과를 비교하려면 다음처럼 한 번 테스트할 수 있습니다.

```bash
docker compose exec airflow airflow dags test \
  hourly_pageviews '2026-10-24T10:00:00+00:00'

```

이 수동 테스트에서 확인한 데이터 구간은 UTC **09:00 이상 ~ 10:00 미만** 입니다. 결과 파일도 09시 디렉터리에 생깁니다.

```bash
docker compose exec airflow cat /tmp/ch04/20261024-09/summary.json

```

```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장을 마치며

처음의 다섯 질문에 답해 보면 이렇습니다.

1. **실행한 시각과 맡은 시간은 다릅니다.** 9~10시를 맡은 실행은 10시가 지나서 시작하므로 `datetime.now()`는 10시 이후를 가리키고, 늦게 돌거나 다시 돌리면 더 멀어집니다. `data_interval_start`(09:00)와 `data_interval_end`(10:00)는 실행이 만들어질 때 정해져 바뀌지 않습니다. `logical_date`는 예약 실행에서는 09:00이지만 날짜를 지정해 직접 실행하면 그 시각이 되므로, 처리할 시간은 데이터 구간에서 꺼냅니다.
2. **구간의 끝은 직접 더하지 않고 `data_interval_end`에서 꺼냅니다.** Airflow가 실행을 만들 때 스케줄을 보고 계산해 둔 값이라, 직접 실행하거나 평일에만 도는 DAG처럼 `logical_date`에 한 시간·하루를 더하면 틀리는 경우에도 맞습니다. 이번 예제는 위키미디어 파일 이름이 구간의 끝 시각이라 파일 주소에 `data_interval_end`를, 결과 폴더에 `data_interval_start`를 썼습니다.
3. **템플릿은 각 태스크를 실행하기 직전에 렌더링합니다.** 오퍼레이터의 `template_fields`에 포함된 속성만 대상이며, Python 함수 본문의 일반 문자열까지 바꾸지는 않습니다. `airflow tasks render`로 채워진 값을 미리 볼 수 있지만, 경로가 렌더링됐다고 파일이 만들어진 것은 아니므로 실행 결과와 나눠 확인합니다.
4. **인자 이름을 보고 콘텍스트 값을 채워 주는 것은 `PythonOperator`입니다.** 인자 이름이 `data_interval_start`면 날짜 객체를 받고, `op_kwargs`의 템플릿으로 넘기면 렌더링된 글자를 받습니다. `op_args`는 순서로, `op_kwargs`는 이름으로 함수 인자에 짝지어집니다.
5. **Connection은 접속 정보, Hook은 외부 시스템 통신, Operator는 태스크의 작업을 맡습니다.** 현재 프로바이더에서 지원하는 오퍼레이터와 연결 ID를 사용합니다.

3장에서 “이번 실행이 어느 시간의 데이터를 맡는가”를 정했다면, 4장에서는 **그 시간을 URL·파일 경로·함수 인자로 전달해 실제 처리 코드와 연결** 했습니다. 같은 DAG 정의를 여러 시간에 사용할 수 있는 이유가 여기에 있습니다.

다음 **5장** (태스크 간 의존성 정의하기)에서는 일렬로 이어진 순서를 넘어, 여러 태스크를 모으고 나누는 의존성, 조건에 따른 분기, 트리거 규칙, XCom과 TaskFlow API를 살펴봅니다.

* * *

_이 글은 Apache Airflow 한국 사용자 모임 운영진으로 활동하며 《Apache Airflow 기반의 데이터 파이프라인》을 학습하고, Airflow 3에서 달라진 내용을 더해 정리한 기록입니다. 글을 읽으시며 의견이나 함께 나누고 싶은 인사이트가 있다면 언제든 댓글로 남겨주세요!_