1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
# ★ 기본 Python 배치 스크립트 구조
import logging
import sys
from datetime import date, timedelta
from contextlib import contextmanager
import psycopg2

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler(f'/var/log/batch/settlement_{date.today()}.log'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

@contextmanager
def get_connection():
    conn = psycopg2.connect(
        host='db-host', dbname='orders', user='batch_user', password='***'
    )
    try:
        yield conn
    finally:
        conn.close()

def run_settlement(target_date: date):
    logger.info(f"정산 배치 시작: target_date={target_date}")

    with get_connection() as conn:
        with conn.cursor() as cur:
            # 1. 데이터 조회
            cur.execute("""
                SELECT seller_id, SUM(amount) as total, COUNT(*) as cnt
                FROM orders
                WHERE order_date = %s AND status = 'COMPLETED'
                GROUP BY seller_id
            """, (target_date,))
            rows = cur.fetchall()
            logger.info(f"대상 판매자 수: {len(rows)}")

            # 2. 정산 결과 저장 (UPSERT — 멱등성)
            for seller_id, total, cnt in rows:
                fee = total * 0.03
                settle_amount = total - fee
                cur.execute("""
                    INSERT INTO settlement (seller_id, settle_date, amount, fee, settle_amount)
                    VALUES (%s, %s, %s, %s, %s)
                    ON CONFLICT (seller_id, settle_date)
                    DO UPDATE SET amount = EXCLUDED.amount,
                                  fee = EXCLUDED.fee,
                                  settle_amount = EXCLUDED.settle_amount
                """, (seller_id, target_date, total, fee, settle_amount))

            conn.commit()
            logger.info(f"정산 배치 완료: {len(rows)}건 처리")

if __name__ == '__main__':
    target = date.today() - timedelta(days=1)
    try:
        run_settlement(target)
    except Exception as e:
        logger.error(f"정산 배치 실패: {e}", exc_info=True)
        sys.exit(1)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# ★ pandas chunksize를 활용한 대용량 ETL
import pandas as pd
from sqlalchemy import create_engine

source_engine = create_engine('postgresql://user:pass@source-db/orders')
target_engine = create_engine('postgresql://user:pass@target-db/analytics')

def etl_daily_orders(target_date: str):
    query = f"""
        SELECT order_id, seller_id, buyer_id, amount, category, order_date
        FROM orders WHERE order_date = '{target_date}'
    """
    total_processed = 0

    for chunk in pd.read_sql(query, source_engine, chunksize=50000):
        # Transform
        chunk['amount_krw'] = chunk['amount'].astype(float)
        chunk['fee'] = chunk['amount_krw'] * 0.03
        chunk['settle_amount'] = chunk['amount_krw'] - chunk['fee']

        result = chunk[['order_id', 'seller_id', 'amount_krw',
                        'fee', 'settle_amount', 'order_date']]

        # Load
        result.to_sql('daily_order_summary', target_engine,
                      if_exists='append', index=False, method='multi',
                      chunksize=5000)

        total_processed += len(chunk)
        logger.info(f"진행: {total_processed}건 처리 완료")

    logger.info(f"ETL 완료: 총 {total_processed}건")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# ★ SQLAlchemy Core — 벌크 UPSERT
from sqlalchemy import create_engine, text
from sqlalchemy.dialects.postgresql import insert

engine = create_engine('postgresql://user:pass@db/app', pool_size=5)

def bulk_upsert_settlements(settlements: list[dict]):
    """대량 UPSERT — PostgreSQL ON CONFLICT 활용"""
    with engine.begin() as conn:
        stmt = insert(settlement_table).values(settlements)
        upsert_stmt = stmt.on_conflict_do_update(
            index_elements=['seller_id', 'settle_date'],
            set_={
                'amount': stmt.excluded.amount,
                'fee': stmt.excluded.fee,
                'settle_amount': stmt.excluded.settle_amount,
                'updated_at': text('NOW()')
            }
        )
        result = conn.execute(upsert_stmt)
        logger.info(f"UPSERT 완료: {result.rowcount}건")

def batch_process_in_chunks(query: str, chunk_size: int = 10000):
    """커서 기반 청크 처리 — 메모리 효율적"""
    with engine.connect() as conn:
        result = conn.execution_options(stream_results=True).execute(text(query))
        while True:
            rows = result.fetchmany(chunk_size)
            if not rows:
                break
            settlements = [transform_to_settlement(row) for row in rows]
            bulk_upsert_settlements(settlements)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
# ★ Generator 패턴 — 대용량 파일 처리
def read_large_csv(filepath: str, chunk_size: int = 10000):
    """메모리를 chunk_size만큼만 사용하는 Generator"""
    import csv
    with open(filepath, 'r') as f:
        reader = csv.DictReader(f)
        chunk = []
        for row in reader:
            chunk.append(row)
            if len(chunk) >= chunk_size:
                yield chunk
                chunk = []
        if chunk:
            yield chunk

for chunk in read_large_csv('/data/orders_20260413.csv'):
    process_and_save(chunk)


# ★ multiprocessing 패턴 — CPU-bound 병렬 처리
from multiprocessing import Pool

def process_partition(args):
    partition_id, start_id, end_id = args
    engine = create_engine('postgresql://...')  # 프로세스별 커넥션
    query = f"SELECT * FROM orders WHERE id BETWEEN {start_id} AND {end_id}"
    df = pd.read_sql(query, engine)
    result = heavy_computation(df)
    result.to_sql('results', engine, if_exists='append', index=False)
    return len(result)

partitions = [(i, i*100000+1, (i+1)*100000) for i in range(8)]
with Pool(processes=8) as pool:
    results = pool.map(process_partition, partitions)
    print(f"총 처리 건수: {sum(results)}")


# ★ asyncio 패턴 — I/O-bound 동시 처리 (API 호출)
import asyncio
import aiohttp

async def batch_api_calls(item_ids: list[int]):
    async with aiohttp.ClientSession() as session:
        semaphore = asyncio.Semaphore(50)  # 동시 요청 제한

        async def limited_fetch(item_id):
            async with semaphore:
                async with session.get(f"{API_URL}/items/{item_id}") as resp:
                    return await resp.json()

        results = await asyncio.gather(
            *[limited_fetch(id) for id in item_ids]
        )
        return results

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
# ★ DAG 정의 예시 — 일일 정산 파이프라인
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['data-team@company.com'],
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id='daily_settlement',
    default_args=default_args,
    description='일일 정산 배치 파이프라인',
    schedule='0 2 * * *',              # 매일 새벽 2시
    start_date=datetime(2026, 1, 1),
    catchup=False,                     # 과거 미실행분 자동 실행 안 함
    tags=['settlement', 'daily'],
) as dag:

    extract = PythonOperator(
        task_id='extract_orders',
        python_callable=extract_orders,
        op_kwargs={'date': '{{ ds }}'},  # 실행 날짜 자동 주입
    )

    validate = PythonOperator(
        task_id='validate_data',
        python_callable=validate_data,
    )

    settle = BashOperator(
        task_id='run_settlement',
        bash_command='python /app/batch/settlement.py --date {{ ds }}',
    )

    notify = SlackWebhookOperator(
        task_id='notify_completion',
        slack_webhook_conn_id='slack_webhook',
        message='✅ {{ ds }} 정산 배치 완료',
    )

    extract >> validate >> settle >> notify

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# ★ XCom 사용 예시
def extract_orders(**context):
    orders = fetch_orders_from_db(context['ds'])
    row_count = len(orders)
    context['ti'].xcom_push(key='row_count', value=row_count)
    context['ti'].xcom_push(key='file_path', value='/data/orders.parquet')
    return row_count  # return 값은 key='return_value'로 자동 저장

def validate_data(**context):
    row_count = context['ti'].xcom_pull(
        task_ids='extract_orders', key='row_count')
    file_path = context['ti'].xcom_pull(
        task_ids='extract_orders', key='file_path')
    if row_count == 0:
        raise ValueError("추출된 데이터가 없습니다!")

# Jinja 템플릿으로도 접근 가능
settle = BashOperator(
    task_id='run_settlement',
    bash_command='python settle.py --count {{ ti.xcom_pull(task_ids="extract_orders", key="row_count") }}',
)

1
2
3
4
5
6
7
8
9
10
# ★ Hook 직접 사용 예시
from airflow.providers.postgres.hooks.postgres import PostgresHook

def extract_with_hook(**context):
    hook = PostgresHook(postgres_conn_id='orders_db')
    df = hook.get_pandas_df(
        "SELECT * FROM orders WHERE order_date = %(date)s",
        parameters={'date': context['ds']}
    )
    return len(df)

1
2
3
4
5
6
7
8
9
10
11
12
# ✓ 멱등한 DAG — 날짜 파라미터 활용
def process_orders(**context):
    execution_date = context['ds']  # '2026-04-13'

    # DELETE-INSERT 패턴으로 멱등성 보장
    hook = PostgresHook('orders_db')
    hook.run(f"DELETE FROM daily_summary WHERE summary_date = '{execution_date}'")
    # ... INSERT ...

# ❌ 안티패턴: datetime.now() 사용
def bad_process():
    today = datetime.now().date()  # 백필 시 잘못된 날짜!

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# ★ 에러 핸들링 종합 예시
def on_failure(context):
    ti = context['ti']
    message = (
        f":red_circle: *배치 실패*\n"
        f"DAG: `{context['dag'].dag_id}`\n"
        f"Task: `{ti.task_id}`\n"
        f"Error: `{context.get('exception', 'Unknown')}`\n"
        f"<{ti.log_url}|로그 확인>"
    )
    send_slack_message(message)

with DAG('settlement_dag', ...) as dag:
    process = PythonOperator(
        task_id='process_settlement',
        python_callable=process_settlement,
        retries=3,
        retry_delay=timedelta(minutes=5),
        retry_exponential_backoff=True,
        max_retry_delay=timedelta(minutes=30),
        on_failure_callback=on_failure,
        execution_timeout=timedelta(hours=2),
    )

    # 성공/실패 무관하게 항상 실행
    cleanup = PythonOperator(
        task_id='cleanup',
        python_callable=cleanup,
        trigger_rule='all_done',
    )

    process >> cleanup

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# ★ SLA 설정 예시
def sla_miss_callback(dag, task_list, blocking_task_list, slas, blocking_tis):
    task_names = [t.task_id for t in task_list]
    send_slack_message(
        f":warning: *SLA 위반!*\n"
        f"DAG: `{dag.dag_id}`\n"
        f"Tasks: {task_names}"
    )

with DAG('daily_settlement', sla_miss_callback=sla_miss_callback, ...) as dag:
    extract = PythonOperator(
        task_id='extract',
        python_callable=extract_fn,
        sla=timedelta(hours=1),  # 1시간 이내 완료되어야 함
    )

1
2
3
4
5
6
7
8
# ❌ 안티패턴: 모듈 레벨에서 무거운 import
import pandas as pd        # DAG 파싱마다 import 발생
import tensorflow as tf

# ✓ 올바른 패턴: 함수 내에서 import
def heavy_processing(**context):
    import pandas as pd    # Task 실행 시에만 import
    df = pd.read_sql(...)

1
2
3
4
5
6
7
task = PythonOperator(
    task_id='heavy_db_query',
    python_callable=query_fn,
    pool='db_connection_pool',
    pool_slots=2,
    priority_weight=10,
)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
# ★ Airflow에서 Spring Batch Job 트리거
from airflow.providers.http.operators.http import SimpleHttpOperator
from airflow.providers.http.sensors.http import HttpSensor

trigger_batch = SimpleHttpOperator(
    task_id='trigger_settlement_batch',
    http_conn_id='spring_batch_api',
    endpoint='/batch/jobs/settlement',
    method='POST',
    data=json.dumps({'date': '{{ ds }}'}),
    headers={'Content-Type': 'application/json'},
    response_filter=lambda resp: json.loads(resp.text)['executionId'],
    do_xcom_push=True,
)

wait_batch = HttpSensor(
    task_id='wait_settlement_complete',
    http_conn_id='spring_batch_api',
    endpoint='/batch/jobs/settlement/status/{{ ti.xcom_pull(task_ids="trigger_settlement_batch") }}',
    response_check=lambda resp: json.loads(resp.text)['status'] == 'COMPLETED',
    poke_interval=60,
    timeout=7200,
    mode='reschedule',
)

trigger_batch >> wait_batch >> downstream_tasks

1
2
3
4
5
6
7
8
9
10
11
12
# ★ Airflow 중복 실행 방지
with DAG(
    'daily_settlement',
    max_active_runs=1,           # ★ DAG 동시 실행 1개
    catchup=False,
    ...
) as dag:
    task = PythonOperator(
        task_id='settle',
        depends_on_past=True,    # ★ 이전 실행 성공 후에만 실행
        ...
    )