데이터 분석 리포트 자동화

src/content/documents/data-analysis/automating-data-analysis-reports.json

매주 월요일 아침, 지난주 매출을 정리해 팀에 공유하는 리포트를 손으로 만드는 분석가가 있다고 하자. 매번 같은 순서 — 데이터를 불러오고, 통계를 계산하고, 표로 정리해 붙여넣는다. 이 반복을 코드 몇 줄로 없앨 수 있다.

한 줄 정의

schedule 로 반복 실행 시각을 예약하고, Jinja2 로 분석 결과를 HTML 리포트로 렌더링하고, 수집·검증·변환·적재를 독립된 함수로 나눈 ETL 파이프라인 으로 이 전체 흐름을 사람 개입 없이 돌아가게 만든다.

왜 자동화하는가

이유

내용

반복 비용 제거

매주 수십 분 걸리던 수작업이 자동화 이후에는 0분이 된다.

최신성

새벽에 완성돼 있어야 하는 리포트는 사람이 출근해서 만들면 이미 늦다.

오류 감소

복사·붙여넣기 실수가 사라지고, 같은 코드는 항상 같은 결과를 낸다.

확장성

지역 1개 리포트를 만드는 코드나 100개를 만드는 코드나 구조는 같다.

schedule로 반복 실행 예약하기

schedule 은 Python 코드 안에서 "언제 무엇을 실행할지"를 사람이 읽는 문장처럼 표현하는 라이브러리다.

import schedule
import time
import logging

def run_daily_report():
    try:
        df = load_and_clean("sales.csv")
        stats = compute_stats(df)
        render_report(stats)
        logging.info("리포트 완료")
    except Exception as e:
        logging.error(f"실패: {e}")

schedule.every().day.at("08:00").do(run_daily_report)
schedule.every().monday.at("09:00").do(weekly_summary)
schedule.every(1).hours.do(check_new_data)

while True:
    schedule.run_pending()
    time.sleep(60)

실제로 몇 초 간격으로 동작하는지 짧게 확인해 본다.

schedule.every(2).seconds.do(lambda: logging.info("리포트 생성 완료"))

start = time.time()
while time.time() - start < 5:
    schedule.run_pending()
    time.sleep(0.5)
2026-08-10 12:25:33 INFO 리포트 생성 완료
2026-08-10 12:25:35 INFO 리포트 생성 완료

schedule 은 스크립트가 실행 중인 동안에만 동작한다. 컴퓨터가 꺼지거나 스크립트가 죽으면 예약도 함께 멈춘다. 서버에 상시 배포하려면 macOS의 launchd 나 Linux의 cron 처럼 OS가 직접 관리하는 스케줄러에 스크립트 실행을 맡기는 것이 더 안전하다. schedule 은 프로토타입이나 로컬 자동화에 적합하다.

Jinja2로 HTML 리포트 만들기

계산까지 끝낸 결과를 사람이 읽는 형태로 바꾸는 단계다. Jinja2는 {{ 변수 }}, {% for %}, {% if %} 로 HTML 안에 값과 반복·조건을 끼워 넣는다.

<!doctype html>
<html>
<body>
  <h1>{{ title }}</h1>
  <p>생성 시각: {{ generated }}</p>
  <ul>
    {% for region, total in summary.items() %}
    <li>{{ region }}: {{ "{:,.0f}".format(total) }}원</li>
    {% endfor %}
  </ul>
  {% if alert %}
  <p style="color:red">주의: {{ alert }}</p>
  {% endif %}
</body>
</html>
from jinja2 import Environment, FileSystemLoader
from pathlib import Path
from datetime import datetime

env = Environment(loader=FileSystemLoader("templates"))
tmpl = env.get_template("report.html")

html = tmpl.render(
    title="월간 판매 분석",
    generated=datetime.now().strftime("%Y-%m-%d"),
    summary={"서울": 2431453, "부산": 1417045, "대구": 1319572, "인천": 1064265},
    alert=None,
)

Path("output/report.html").write_text(html, encoding="utf-8")
<h1>월간 판매 분석</h1>
<p>생성 시각: 2026-08-10</p>
<ul>
  <li>서울: 2,431,453원</li>
  <li>부산: 1,417,045원</li>
  <li>대구: 1,319,572원</li>
  <li>인천: 1,064,265원</li>
</ul>

템플릿 파일은 계산 코드와 분리돼 있어, 디자인만 바꿀 때 Python 코드를 건드릴 필요가 없다. 완성된 HTML을 webbrowser.open() 으로 바로 열거나, pdfkit 같은 라이브러리로 PDF로 변환해 첨부 파일로 보낼 수도 있다.

ETL 파이프라인 원칙 — 단계를 분리한다

좋은 파이프라인은 네 가지를 만족한다: 각 단계가 분리되어 있고, 실패가 기록되고, 재현 가능하고, 단계별로 테스트 가능하다. 데이터 수집·검증·집계·저장을 한 함수에 몰아넣지 않고 나눈다.

import logging
from pydantic import BaseModel, ValidationError

logger = logging.getLogger("pipeline")


class SalesRecord(BaseModel):
    region: str
    amount: float


def extract(rows: list[dict]) -> list[dict]:
    logger.info(f"수집 완료: {len(rows)}건")
    return rows


def validate(rows: list[dict]):
    ok, errors = [], []
    for row in rows:
        try:
            ok.append(SalesRecord(**row))
        except ValidationError as e:
            errors.append((row, str(e)))
    if errors:
        logger.warning(f"검증 오류: {len(errors)}건")
    return ok, errors


def transform(records: list[SalesRecord]) -> dict:
    return {"total": sum(r.amount for r in records), "count": len(records)}


def load(summary: dict) -> None:
    logger.info(f"적재 완료: {summary}")


def run_pipeline(raw_rows: list[dict]) -> dict:
    logger.info("파이프라인 시작")
    raw = extract(raw_rows)
    validated, errors = validate(raw)
    summary = transform(validated)
    load(summary)
    logger.info("파이프라인 종료")
    return summary

잘못된 값이 섞인 입력으로 실제 실행한 결과다.

rows = [
    {"region": "서울", "amount": 12000},
    {"region": "부산", "amount": "안됨"},   # 잘못된 값
    {"region": "대구", "amount": 8000},
]
run_pipeline(rows)
INFO 파이프라인 시작
INFO 수집 완료: 3건
WARNING 검증 오류: 1건
INFO 적재 완료: {'total': 20000.0, 'count': 2}
INFO 파이프라인 종료

검증 오류가 있는 행 하나 때문에 전체 파이프라인이 멈추지 않는다. 문제가 있는 행은 로그에 남기고, 나머지 두 건으로 집계를 이어간다. 각 함수(extract, validate, transform, load)는 독립적이라 22편에서 다룬 방식대로 각각 pytest로 따로 테스트할 수 있고, 실패했을 때 로그만 봐도 어느 단계에서 멈췄는지 바로 알 수 있다.

장점과 한계

구분

내용

장점

반복 작업의 사람 개입을 없애 시간과 실수를 동시에 줄이고, 각 단계가 분리돼 있어 문제 지점을 빠르게 좁힐 수 있다.

한계

자동화된 파이프라인은 실패해도 사람이 못 볼 수 있다. 로깅만으로는 부족하고, 실패를 알림(이메일·슬랙 등)으로 연결하는 모니터링이 함께 필요하다.

자주 발생하는 문제

schedule.run_pending()을 반복문 없이 한 번만 호출한다. run_pending() 은 그 순간 실행 시각이 된 작업만 처리하고 즉시 반환한다. 무한 반복문 안에서 주기적으로 불러줘야 실제로 예약된 시각마다 실행된다.

Jinja2 템플릿에 외부 입력을 그대로 꽂는다. 기본 Environment 는 값을 자동으로 이스케이프하지 않는다. 사용자 입력이 섞인 값을 HTML 리포트에 그대로 넣으면 스크립트가 그대로 실행될 수 있다.

env = Environment(loader=FileSystemLoader("templates"))  # autoescape 꺼짐
tmpl = env.from_string("<p>{{ name }}</p>")
tmpl.render(name="<script>alert(1)</script>")
# '<p><script>alert(1)</script></p>'  그대로 삽입된다
from jinja2 import select_autoescape

env = Environment(loader=FileSystemLoader("templates"), autoescape=select_autoescape())
tmpl = env.from_string("<p>{{ name }}</p>")
tmpl.render(name="<script>alert(1)</script>")
# '<p>&lt;script&gt;alert(1)&lt;/script&gt;</p>'  안전하게 이스케이프됨

외부에서 온 값이 들어가는 템플릿에는 항상 autoescape=select_autoescape() 를 켠다.

파이프라인 실패를 로그로만 남기고 아무도 확인하지 않는다. logging.error 는 파일이나 콘솔에 기록될 뿐, 사람에게 알림을 보내지 않는다. 리포트가 며칠째 갱신되지 않았는데 아무도 몰랐다면 로그가 아니라 알림 채널이 빠진 것이다.

관련 문서

이 파이프라인의 검증 단계는 Pydantic으로 외부 데이터를 검증하기 의 모델을 그대로 재사용한다. 로그 설계 원칙은 print 대신 logging을 사용해야 하는 이유 를 먼저 본다.

참고 자료

댓글 0

댓글을 불러오는 중…