June 8, 2025docker-compose.yaml 파일에 컨테이너들의 설정을 입력.yaml 파일이 있는 위치에서 docker compose up 명령어를 입력하여 실행x-airflow-common: # 각 서비스에 공통 적용될 항목들
services: # 컨테이너로 실행할 서비스를 정의
volumns: # 컨테이너에 연결할 볼륨을 정의
networks: # 컨테이너에 연결할 네트워크를 정의x-airflow-common
June 7, 2025HttpOperator 는 Provider 패키지로 별도의 설치가 필요apache-airflow>=2.10.0 버전을 요구pip install apache-airflow-providers-httphttp_conn_id : http Connection을 생성해야 하는데 해당 Connection의 IDendpoint : Connection에 등록한 Host 뒤에 붙는 경로method : HTTP 메서드 (GET, POST, PUT, DELETE 등)data : POST 요청 시 전달할 데이터 또는 GET 요청 시 전달할 파라미터headers : HTTP 요청 헤더response_check : HTTP 응답 결과가 정상인지 확인하는 함수 (True 반환)response_filter : HTTP 응답 결과에 대한 전처리 함수검색 > 쇼핑 쇼핑 검색 개요 개요 사전 준비 사항 쇼핑 검색 API 레퍼런스 쇼핑 검색 결과 조회 오류 코드 검색 API 쇼핑 검색 구현 예제 쇼핑 검색 개요 개요 사전 준비 사항 개요 검색 API와 쇼핑 검색 개 …
June 7, 2025
| 옵션 | 설명 |
|---|---|
all_success | 기본값, 상위 Task가 모두 성공하면 실행 |
all_failed | 상위 Task가 모두 failed 상태면 실행 |
all_done | 상위 Task가 모두 수행되면 실행 (성공 또는 실패) |
all_skipped | 상위 Task가 모두 skipped 상태면 실행 |
one_failed | 상위 Task 중 하나 이상 실패하면 실행 |
one_success | 상위 Task 중 하나 이상 성공하면 실행 |
one_done | 상위 Task 중 하나 이상 수행되면 실행 (성공 또는 실패) |
none_failed | 상위 Task 중에 failed 상태가 없으면 실행 |
none_failed_min_one_success | 상위 Task 중에 failed 상태가 없고 성공한 Task가 1개 이상이면 실행 |
none_skipped | 상위 Task 중에 skipped 상태가 없으면 실행 |
always | 항상 실행 |
all_done 의 동작을 확인하기 위한 예시 DAG 작성failed 상태를 유발downstream_task 에 trigger_rule 파라미터로 all_done 전달# dags/trigger_rule1.py
from airflow.sdk import DAG, task
from airflow.exceptions import AirflowException
from airflow.providers.standard.operators.bash import BashOperator
import pendulum
with DAG(
dag_id="trigger_rule1",
start_date=pendulum.datetime(2025, 1, 1, tz="Asia/Seoul"),
schedule="0 0 * * *",
catchup=False,
tags=["example", "branch"],
) as dag:
upstream_task1 = BashOperator(
task_id="upstream_task1",
bash_command="echo upstream1"
)
@task(task_id="upstream_task2")
def upstream_task2():
raise AirflowException("upstream2 Exception")
@task(task_id="upstream_task3")
def upstream_task3():
print("정상 처리")
@task(task_id="downstream_task", trigger_rule="all_done")
def downstream_task():
print("정상 처리")
[upstream_task1, upstream_task2(), upstream_task3()] >> downstream_task()all_done 은 상위 Task가 성공 또는 실패 여부에 관계없이 모두 수행되면 실행하는 옵션으로,
upstream_task2 가 실패 처리되어도 downstream_task 가 수행되는 모습을 확인
none_skipped 의 동작을 확인하기 위한 예시 DAG 작성skipped 상태를 유발downstream_task 에 trigger_rule 파라미터로 none_skipped 전달# dags/trigger_rule2.py
from airflow.sdk import DAG, task
from airflow.providers.standard.operators.bash import BashOperator
import pendulum
with DAG(
dag_id="trigger_rule2",
start_date=pendulum.datetime(2025, 1, 1, tz="Asia/Seoul"),
schedule="0 0 * * *",
catchup=False,
tags=["example", "branch"],
) as dag:
@task.branch(task_id="branching")
def random_branch():
import random
item_lst = ['A','B','C']
selected_item = random.choice(item_lst)
if selected_item == 'A':
return "upstream_task_a"
elif selected_item == 'B':
return "upstream_task_b"
elif selected_item == 'C':
return "upstream_task_c"
upstream_task_a = BashOperator(
task_id="upstream_task_a",
bash_command="echo upstream1"
)
@task(task_id="upstream_task_b")
def upstream_task_b():
print("정상 처리")
@task(task_id="upstream_task_c")
def upstream_task_c():
print("정상 처리")
@task(task_id="downstream_task", trigger_rule="none_skipped")
def downstream_task():
print("정상 처리")
random_branch() >> [upstream_task_a, upstream_task_b(), upstream_task_c()] >> downstream_task()none_skipped 은 상위 Task가 skipped 상태가 아니어야 실행하는 옵션으로, upstream_task1 만 성공하고
나머지는 skipped 처리되었기 때문에, downstream_task 도 수행되지 못하고 skipped 처리
TriggerDagRunOperator 를 사용하고, 선행 DAG이 2개 이상인 경우는 ExternalTaskSensor 를 사용 권장
scheduled__2025-06-01T00:00:00+00:00trigger_run_id : DAG을 실행시킬 때 어떤 run_id 로 실행할지 지정 가능logical_date : DAG이 트리거된 시간을 지정 가능, manual__{{logical_date}}reset_dag_run : run_id 로 수행된 이력이 있어도 실행시키려면 True 로 설정wait_for_completion : 지정한 DAG이 완료되어야 다음 Task를 실행하고 싶을 경우 True 로 설정success 로 빠져나가 다음 Task를 실행poke_interval : 지정한 DAG이 완료되었는지 확인하는 주기allowed_states : Task가 success 상태가 되기 위한 DAG의 처리 상태 목록failed_states : Task가 failed 상태가 되기 위한 DAG의 처리 상태 목록# dags/trigger_dagrun.py
from airflow.sdk import DAG
from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.providers.standard.operators.bash import BashOperator
import pendulum
with DAG(
dag_id="trigger_dagrun",
start_date=pendulum.datetime(2025, 1, 1, tz="Asia/Seoul"),
schedule="0 0 * * *",
catchup=False,
tags=["example", "dagrun"],
) as dag:
start_task = BashOperator(
task_id="start_task",
bash_command="echo \"start!\"",
)
trigger_dag_task = TriggerDagRunOperator(
task_id="trigger_dag_task",
trigger_dag_id="python_operator",
trigger_run_id=None,
logical_date="{{data_interval_start}}",
reset_dag_run=True,
wait_for_completion=False,
poke_interval=60,
allowed_states=["success"],
failed_states=None
)
start_task >> trigger_dag_tasktrigger_dag_task 의 실행 로그에서 python_operator 가 호출된 것을 확인[2025-06-07, 10:52:54] INFO - DAG bundles loaded: dags-folder: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-06-07, 10:52:54] INFO - Filling up the DagBag from /opt/airflow/dags/trigger_dagrun.py: source="airflow.models.dagbag.DagBag"
[2025-06-07, 10:52:54] INFO - Triggering Dag Run.: trigger_dag_id="python_operator": source="task"
[2025-06-07, 10:52:54] INFO - Dag Run triggered successfully.: trigger_dag_id="python_operator": source="task"run_id 가 첫 번째 이미지인 TriggerDagRunOperator의 실행 시간과 같다는 것을 알 수 있으며,
trigger_run_id 를 지정하지 않았기 때문에 manual 로 지정
June 6, 2025
BranchPythonOperator 에서 랜덤한 조건에 따라 task_a 만 수행하거나, task_b 와 task_c 를 같이 수행하는 분기 처리# dags/branch_python.py
from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator
from airflow.providers.standard.operators.python import BranchPythonOperator
import pendulum
with DAG(
dag_id="branch_python",
start_date=pendulum.datetime(2025, 1, 1, tz="Asia/Seoul"),
schedule="0 0 * * *",
catchup=False,
tags=["example", "branch"],
) as dag:
def select_random():
import random
item_lst = ['A','B','C']
selected_item = random.choice(item_lst)
if selected_item == 'A':
return "task_a"
else:
return ["task_b","task_c"]
branch_task = BranchPythonOperator(
task_id="branch_task",
python_callable=select_random
)
def print_selected(**kwargs):
print(kwargs["selected"])
task_a = PythonOperator(
task_id="task_a",
python_callable=print_selected,
op_kwargs={"selected":'A'}
)
task_b = PythonOperator(
task_id="task_b",
python_callable=print_selected,
op_kwargs={"selected":'B'}
)
task_c = PythonOperator(
task_id="task_c",
python_callable=print_selected,
op_kwargs={"selected":'C'}
)
branch_task >> [task_a, task_b, task_c]task_a 만 수행되거나, task_b 와 task_c 가 같이 수행되는 두 가지 경우를 확인
task_a 가 선택되는 작업에서 XCom을 보면 skipmixin_key 키로 {'followed': ['task_a']} 값이 전달되는데,
이를 통해 다른 Task에서도 어떤 분기 처리가 되었는지 확인 가능![skipmixin_key = {'followed': ['task_a']}](https://dl.dropboxusercontent.com/scl/fi/jrraft65v5seeqr3hr21f/airflow-28-branch-skipmixin-key.webp?rlkey=ndgnddwg7dxwruzts4ke8pywa&dl=0)
June 1, 2025{{ }} 를 이용해 변수를 치환된 값으로 입력data_interval_start : 스케줄의 시작 날짜이며,pendulum.DateTime 타입data_interval_end : 스케줄의 종료 날짜(= 배치일)이며,pendulum.DateTime 타입logical_date : DAG가 실행 중인 시점의 날짜이며,pendulum.DateTime 타입ds : logical_date 를 YYYY-MM-DD 형태의 문자열로 변환한 값ds 에서 - 을 제거한 YYYYMMDD 형태의 문자열 ds_nodash 변수도 제공ts : logical_date 를 2018-01-01T00:00:00+00:00 형태의 문자열로 변환한 값ts_nodash_with_tz 또는 ts_nodash 등의 변형된 변수도 지원ds 또는 ts 등은 {{ logical_date | ds }} 의 형태로도 표현 가능bash_command, env 파라미터에 템플릿 적용 가능templates_dict, op_args, op_kwargs 파라미터에 템플릿 적용 가능env로 파라미터를 전달해서 출력# dags/bash_template.py
from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator
import pendulum
with DAG(
dag_id="bash_template",
schedule="0 0 * * *",
start_date=pendulum.datetime(2025, 1, 1, tz="Asia/Seoul"),
catchup=False,
tags=["example", "template"],
) as dag:
bash_task1 = BashOperator(
task_id="bash_task1",
bash_command="echo \"End date is {{ data_interval_end }}\"",
)
bash_task2 = BashOperator(
task_id="bash_task2",
env={
"START_DATE": "{{ data_interval_start | ds }}",
"END_DATE": "{{ data_interval_end | ds }}"
},
bash_command="echo \"Start date is $START_DATE \" && echo \"End date is $END_DATE\"",
)
bash_task1 >> bash_task2bash_task1 의 실행 로그에서 data_interval_end 가 시간대를 포함하여 전체 출력된 것을 확인bash_task2 의 실행 로그에서는 data_interval_start 와 data_interval_end 이 YYYY-MM-DD 형태의 문자열로 출력된 것을 확인# bash_task1
[2025-06-01, 19:59:27] INFO - DAG bundles loaded: dags-folder: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-06-01, 19:59:27] INFO - Filling up the DagBag from /opt/airflow/dags/bash_template.py: source="airflow.models.dagbag.DagBag"
[2025-06-01, 19:59:27] INFO - Running command: ['/usr/bin/bash', '-c', 'echo "End date is 2025-05-31 15:00:00+00:00"']: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-06-01, 19:59:27] INFO - Output:: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-06-01, 19:59:27] INFO - End date is 2025-05-31 15:00:00+00:00: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-06-01, 19:59:27] INFO - Command exited with return code 0: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-06-01, 19:59:27] INFO - Pushing [2025-06-01, 19:59:28] INFO - DAG bundles loaded: dags-folder: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-06-01, 19:59:28] INFO - Filling up the DagBag from /opt/airflow/dags/bash_template.py: source="airflow.models.dagbag.DagBag"
[2025-06-01, 19:59:28] INFO - Running command: ['/usr/bin/bash', '-c', 'echo "Start date is $START_DATE " && echo "End date is $END_DATE"']: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-06-01, 19:59:28] INFO - Output:: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-06-01, 19:59:28] INFO - Start date is 2025-05-31: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-06-01, 19:59:28] INFO - End date is 2025-05-31: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-06-01, 19:59:28] INFO - Command exited with return code 0: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-06-01, 19:59:28] INFO - Pushing **kwargs 내용을 출력한적이 있었는데, 직접 전달하지 않았음에도 출력되었던 값들이 바로 Jinja 템플릿 변수에 해당# dags/python_template1.py
from airflow.sdk import DAG, task
import pendulum
with DAG(
dag_id="python_template1",
schedule="0 0 * * *",
start_date=pendulum.datetime(2025, 1, 1, tz="Asia/Seoul"),
catchup=False,
tags=["example", "template"],
) as dag:
@task(task_id="python_task")
def show_templates(**kwargs):
from pprint import pprint
pprint(kwargs)
show_templates()python_task 의 실행 로그에서 data_interval_end, data_interval_start 등 Jinja 템플릿 변수가 출력된 것을 확인[2025-06-01, 20:18:19] INFO - DAG bundles loaded: dags-folder: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-06-01, 20:18:19] INFO - Filling up the DagBag from /opt/airflow/dags/python_template.py: source="airflow.models.dagbag.DagBag"
[2025-06-01, 20:18:19] INFO - {'conn': <ConnectionAccessor (dynamic access)>,: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'dag': <DAG: python_template>,: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'dag_run': DagRun(dag_id='python_template', run_id='scheduled__2025-05-31T15:00:00+00:00', logical_date=datetime.datetime(2025, 5, 31, 15, 0, tzinfo=TzInfo(UTC)), data_interval_start=datetime.datetime(2025, 5, 31, 15, 0, tzinfo=TzInfo(UTC)), data_interval_end=datetime.datetime(2025, 5, 31, 15, 0, tzinfo=TzInfo(UTC)), run_after=datetime.datetime(2025, 5, 31, 15, 0, tzinfo=TzInfo(UTC)), start_date=datetime.datetime(2025, 6, 1, 11, 18, 19, 250768, tzinfo=TzInfo(UTC)), end_date=None, clear_number=0, run_type=<DagRunType.SCHEDULED: 'scheduled'>, conf={}, consumed_asset_events=[]),: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'data_interval_end': DateTime(2025, 5, 31, 15, 0, 0, tzinfo=Timezone('UTC')),: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'data_interval_start': DateTime(2025, 5, 31, 15, 0, 0, tzinfo=Timezone('UTC')),: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'ds': '2025-05-31',: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'ds_nodash': '20250531',: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'inlet_events': InletEventsAccessors(_inlets=[], _assets={}, _asset_aliases={}),: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'inlets': [],: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'logical_date': DateTime(2025, 5, 31, 15, 0, 0, tzinfo=Timezone('UTC')),: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'macros': <MacrosAccessor (dynamic access to macros)>,: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'map_index_template': None,: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'outlet_events': <airflow.sdk.execution_time.context.OutletEventAccessors object at 0xffffaacaa810>,: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'outlets': [],: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'params': {},: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'prev_data_interval_end_success': <Proxy at 0xffffaad2b140 with factory <function RuntimeTaskInstance.get_template_context.<locals>.<lambda> at 0xffffaad3cfe0>>,: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - Done. Returned value was: None: source="airflow.task.operators.airflow.providers.standard.decorators.python._PythonDecoratedOperator"
[2025-06-01, 20:18:19] INFO - 'prev_data_interval_start_success': <Proxy at 0xffffaad2b0b0 with factory <function RuntimeTaskInstance.get_template_context.<locals>.<lambda> at 0xffffaad3cea0>>,: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'prev_end_date_success': <Proxy at 0xffffaacde870 with factory <function RuntimeTaskInstance.get_template_context.<locals>.<lambda> at 0xffffaad0c7c0>>,: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'prev_start_date_success': <Proxy at 0xffffaacde8d0 with factory <function RuntimeTaskInstance.get_template_context.<locals>.<lambda> at 0xffffaad0c720>>,: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'run_id': 'scheduled__2025-05-31T15:00:00+00:00',: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'task': <Task(_PythonDecoratedOperator): python_task>,: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'task_instance': RuntimeTaskInstance(id=UUID('01972b36-c56a-7d54-b52e-45bb8feb6594'), task_id='python_task', dag_id='python_template', run_id='scheduled__2025-05-31T15:00:00+00:00', try_number=1, map_index=-1, hostname='f6f932b48199', context_carrier={}, task=<Task(_PythonDecoratedOperator): python_task>, bundle_instance=LocalDagBundle(name=dags-folder), max_tries=0, start_date=datetime.datetime(2025, 6, 1, 11, 18, 19, 315990, tzinfo=TzInfo(UTC)), end_date=None, state=<TaskInstanceState.RUNNING: 'running'>, is_mapped=False, rendered_map_index=None),: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'task_instance_key_str': 'python_template__python_task__20250531',: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'task_reschedule_count': 0,: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'templates_dict': None,: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'ti': RuntimeTaskInstance(id=UUID('01972b36-c56a-7d54-b52e-45bb8feb6594'), task_id='python_task', dag_id='python_template', run_id='scheduled__2025-05-31T15:00:00+00:00', try_number=1, map_index=-1, hostname='f6f932b48199', context_carrier={}, task=<Task(_PythonDecoratedOperator): python_task>, bundle_instance=LocalDagBundle(name=dags-folder), max_tries=0, start_date=datetime.datetime(2025, 6, 1, 11, 18, 19, 315990, tzinfo=TzInfo(UTC)), end_date=None, state=<TaskInstanceState.RUNNING: 'running'>, is_mapped=False, rendered_map_index=None),: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'triggering_asset_events': TriggeringAssetEventsAccessor(_events=defaultdict(<class 'list'>, {})),: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'ts': '2025-05-31T15:00:00+00:00',: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'ts_nodash': '20250531T150000',: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'ts_nodash_with_tz': '20250531T150000+0000',: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'var': {'json': <VariableAccessor (dynamic access)>,: chan="stdout": source="task"
[2025-06-01, 20:18:19] INFO - 'value': <VariableAccessor (dynamic access)>}}: chan="stdout": source="task"python_task1 정의python_task2 정의# dags/python_template2.py
from airflow.sdk import DAG, task
import pendulum
with DAG(
dag_id="python_template2",
schedule="0 0 * * *",
start_date=pendulum.datetime(2025, 1, 1, tz="Asia/Seoul"),
catchup=False,
tags=["example", "template"],
) as dag:
def print_period(start_date, end_date, **kwargs):
print(start_date)
print(end_date)
python_task1 = PythonOperator(
task_id="python_task1",
python_callable=print_period,
op_kwargs={
"start_date": "{{ data_interval_start | ds }}",
"end_date": "{{ data_interval_end | ds }}"
},
)
@task(task_id="python_task2")
def python_task2(**kwargs):
for __key in ["ds", "ts", "data_interval_start", "data_interval_end"]:
if __key in kwargs:
print(f"{__key}: {kwargs[__key]}")
python_task1 >> python_task2()python_task1 의 실행 로그에서 전달한 data_interval_start, data_interval_end 값이 YYYY-MM-DD 형태의 문자열로 출력된 것을 확인python_task2 의 실행 로그에서는 ds, ts, data_interval_start, data_interval_end 값을 keyword argument로부터 꺼내서 그대로 출력# python_task1
[2025-06-02, 00:12:27] INFO - DAG bundles loaded: dags-folder: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-06-02, 00:12:27] INFO - Filling up the DagBag from /opt/airflow/dags/python_template2.py: source="airflow.models.dagbag.DagBag"
[2025-06-02, 00:12:27] INFO - Done. Returned value was: None: source="airflow.task.operators.airflow.providers.standard.operators.python.PythonOperator"
[2025-06-02, 00:12:27] INFO - 2025-06-01: chan="stdout": source="task"
[2025-06-02, 00:12:27] INFO - 2025-06-01: chan="stdout": source="task"# python_task2
[2025-06-02, 00:12:27] INFO - DAG bundles loaded: dags-folder: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-06-02, 00:12:27] INFO - Filling up the DagBag from /opt/airflow/dags/python_template2.py: source="airflow.models.dagbag.DagBag"
[2025-06-02, 00:12:27] INFO - Done. Returned value was: None: source="airflow.task.operators.airflow.providers.standard.decorators.python._PythonDecoratedOperator"
[2025-06-02, 00:12:27] INFO - ds: 2025-06-01: chan="stdout": source="task"
[2025-06-02, 00:12:27] INFO - ts: 2025-06-01T15:00:00+00:00: chan="stdout": source="task"
[2025-06-02, 00:12:27] INFO - data_interval_start: 2025-06-01 15:00:00+00:00: chan="stdout": source="task"
[2025-06-02, 00:12:27] INFO - data_interval_end: 2025-06-01 15:00:00+00:00: chan="stdout": source="task"macros.datetime : datetime.datetime 라이브러리 기반 연산 제공macros.timedelta : datetime.timedelta 라이브러리 기반 연산 제공macros.dateutil : dateutil 라이브러리 기반 연산 제공import datetime as dt
from dateutil.relativedelta import relativedelta
today = dt.date.today()
# 1일로 변경
first_date = today.replace(day=1) # datetime 연산
first_date = today + relativedelta(day=1) # dateutil 연산
# 1일 빼기
yesterday = today - dt.timedela(days=1) # timedela 연산
yesterday = today - relativedelta(days=1) # dateutil 연산bash_macros1 은 매월 말에 실행되도록 스케줄을 지정하고,START_DATE, 배치일을 END_DATE 로 설정하여 1일부터 말일이 출력되기를 기대# dags/bash_macros1.py
from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator
import pendulum
with DAG(
dag_id="bash_macros1",
schedule="0 0 L * *",
start_date=pendulum.datetime(2025, 1, 1, tz="Asia/Seoul"),
catchup=False,
tags=["example", "template"],
) as dag:
bash_task1 = BashOperator(
task_id="bash_task1",
env={
"START_DATE": "{{ (data_interval_start.in_timezone(\"Asia/Seoul\") + macros.dateutil.relativedelta.relativedelta(days=1)) | ds }}",
"END_DATE": "{{ data_interval_end.in_timezone(\"Asia/Seoul\") | ds }}"
},
bash_command="echo \"Start date is $START_DATE \" && echo \"End date is $END_DATE\"",
)
bash_task1bash_macros2 는 매월 둘째주 토요일에 실행되도록 스케줄을 지정하고,
직전 배치일과 배치일을 1일로 변경해서 전월 1일과 당월 1일이 출력되기를 기대# dags/bash_macros2.py
from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator
import pendulum
with DAG(
dag_id="bash_macros2",
schedule="0 0 * * 6#2",
start_date=pendulum.datetime(2025, 1, 1, tz="Asia/Seoul"),
catchup=False,
tags=["example", "template"],
) as dag:
bash_task2 = BashOperator(
task_id="bash_task2",
env={
"START_DATE": "{{ (data_interval_end.in_timezone(\"Asia/Seoul\") + macros.dateutil.relativedelta.relativedelta(day=1)) | ds }}",
"END_DATE": "{{ (data_interval_end.in_timezone(\"Asia/Seoul\") + macros.dateutil.relativedelta.relativedelta(day=1)) | ds }}"
},
bash_command="echo \"Start date is $START_DATE \" && echo \"End date is $END_DATE\"",
)
bash_task2data_interval_start, data_interval_end 가 동일한 값을 가지고 있다고 생각됨# bash_task1 (bash_macros1)
[2025-06-03, 10:54:43] INFO - Start date is 2025-06-01: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-06-03, 10:54:43] INFO - End date is 2025-05-31: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"# bash_task2 (bash_macros2)
[2025-06-03, 10:57:23] INFO - Start date is 2025-05-01: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"
[2025-06-03, 10:57:23] INFO - End date is 2025-05-01: source="airflow.task.hooks.airflow.providers.standard.hooks.subprocess.SubprocessHook"bash_macros1 의 마지막 실행 내역에 대해 Details 탭에서 실행 정보를 조회했을 때,
data_interval_start, data_interval_end 값이 모두 동일한 배치일로 나타나는 것을 확인data_interval_start, data_interval_end 를 결정하는 기준이 변경된 것을 인지
May 30, 2025% git init
% git add .gitignore
% git commit -m "init"
% git remote add origin https://github.com/<username>/<repository>
% git push.gitignore 를 로컬 Airflow 설치 경로에 복제하여 불필요한 /logs 경로 등을 Git에서 제외Apache Airflow - A platform to programmatically author, schedule, and monitor workflows - airflow/.gitignore at main · …
May 27, 2025>> 또는 << 연산자를 사용해 선언first_task >> second_task