자동화는 실행보다 운영이 중요하다
앞 장들에서는 엑셀 파일을 정리하고, 데이터베이스를 조회하고, API와 이메일, 브라우저 작업을 자동화하는 방법을 다뤘다. 여기까지는 대부분 사용자가 직접 명령을 실행하는 방식이었다. 예를 들어 터미널에서 python src/daily_report.py를 입력하면 보고서가 만들어지는 식이다.
하지만 실무 자동화의 가치는 한 번 성공한 코드가 아니라 매일 정해진 시간에 안정적으로 실행되는 구조에서 나온다. 담당자가 휴가 중이어도 월요일 오전 8시에 보고서가 만들어져야 한다. API가 잠시 실패하면 몇 분 뒤 다시 시도해야 한다. 데이터 형식이 바뀌면 잘못된 보고서를 보내는 대신 멈추고 담당자에게 알려야 한다.
이 장의 핵심은 개별 스크립트를 업무 프로세스로 발전시키는 것이다. 이를 위해 두 가지 개념을 구분하자.
스케줄링: 언제 실행할지 정하는 일이다. 매일 오전 8시, 매시간, 매월 1일 같은 조건을 지정한다.
워크플로 오케스트레이션: 여러 작업의 순서, 의존성, 재시도, 실행 이력, 알림을 관리하는 일이다.
작은 자동화는 스케줄링만으로 충분할 수 있다. 하지만 데이터 수집, 전처리, 보고서 생성, 알림 발송처럼 여러 단계가 연결되면 오케스트레이션이 필요해진다.
AI 생성 이미지
실행 방식 선택하기
자동화 실행 방식은 업무 규모와 실패 위험에 따라 선택한다. 처음부터 복잡한 도구를 도입할 필요는 없다. 반대로 중요한 업무를 개인 PC의 예약 작업에만 의존하는 것도 위험하다.
| 방식 | 적합한 상황 | 장점 | 주의할 점 |
|---|---|---|---|
| cron | 리눅스 서버에서 간단한 정기 작업 실행 | 가볍고 안정적이며 설정이 단순함 | 작업 순서, 재시도, 실행 상태 확인은 직접 구현해야 함 |
| Windows 작업 스케줄러 | 윈도우 PC나 사내 윈도우 서버에서 실행 | 별도 도구 없이 사용 가능 | PC 절전, 계정 권한, 경로 문제를 주의해야 함 |
| Prefect | 파이썬 스크립트를 비교적 쉽게 워크플로로 확장 | flow, task 단위로 나누고 재시도와 상태 추적을 붙이기 쉬움 | 운영 서버와 대시보드 관리 방식이 필요함 |
| Airflow | 여러 팀이 사용하는 대규모 데이터 파이프라인 | DAG 기반 의존성 관리, 웹 UI, 다양한 연동이 강함 | 초기 설정과 운영 부담이 상대적으로 큼 |
cron은 단순한 정기 실행에 적합하다. 예를 들어 매주 평일 오전 8시에 보고서를 만드는 작업은 다음처럼 등록할 수 있다.
0 8 * * 1-5 cd /app/business-automation && /app/venv/bin/python src/daily_report.py >> logs/cron.log 2>&1이 한 줄은 평일 오전 8시에 프로젝트 폴더로 이동한 뒤 파이썬 스크립트를 실행하고, 출력과 오류를 로그 파일에 남긴다는 뜻이다. 간단하지만 강력하다. 다만 daily_report.py 내부에서 실패 처리, 중복 실행 방지, 알림 발송을 직접 구현해야 한다. 서버 시간대와 환경 변수도 반드시 확인해야 한다. 운영 서버의 시간이 한국 시간이 아니면 오전 8시에 실행한다고 생각한 작업이 전혀 다른 시간에 실행될 수 있다.
Prefect는 기존 파이썬 함수를 task와 flow로 감싸 워크플로로 만드는 방식에 가깝다. 공식 설명에서도 파이썬 스크립트를 운영 워크플로로 끌어올리는 도구로 소개하며, 재시도, 의존성, 스케줄링, 실행 상태 추적을 주요 기능으로 제공한다1.
Airflow는 더 큰 조직과 복잡한 데이터 파이프라인에 적합하다. 작업 흐름을 파이썬으로 정의하고, 웹 애플리케이션에서 실행 상태와 로그를 확인할 수 있으며, 클라우드와 데이터 도구 연동을 위한 다양한 구성 요소를 제공한다2. 이미 회사에 데이터 플랫폼 팀이 있고 여러 부서의 배치 작업을 한곳에서 관리해야 한다면 Airflow가 좋은 선택이 될 수 있다.
선택 기준은 단순하다.
작업 단계가 단순하고 실패해도 영향이 작으면
cron이나 작업 스케줄러로 시작한다.작업이 여러 단계이고 재시도와 상태 확인이 중요하면 Prefect 같은 워크플로 도구를 검토한다.
여러 팀, 수십 개 이상의 파이프라인, 중앙 관리가 필요하면 Airflow를 검토한다.
업무 파이프라인으로 나누기
운영 가능한 자동화는 하나의 긴 코드가 아니라 여러 단계로 나뉜 파이프라인이다. 예를 들어 매일 매출 보고서를 만드는 자동화는 다음 흐름으로 나눌 수 있다.
데이터 수집 -> 검증·전처리 -> 데이터베이스 저장 -> 보고서 생성 -> 알림 발송각 단계는 역할이 분명해야 한다.
데이터 수집: API, 데이터베이스, 엑셀, 웹 다운로드에서 원천 데이터를 가져온다.
검증·전처리: 필수 열, 날짜, 금액, 중복 데이터를 확인한다.
저장: 정리된 데이터를 데이터베이스나 표준 파일 경로에 저장한다.
보고서 생성: 엑셀, PDF, 대시보드용 CSV 같은 결과물을 만든다.
알림 발송: Slack, Teams, 이메일로 결과 위치와 주요 지표를 전달한다.
중요한 점은 뒤 단계가 앞 단계의 성공을 전제로 해야 한다는 것이다. 데이터 검증이 실패했는데 보고서를 만들면 안 된다. 보고서 생성이 실패했는데 성공 알림을 보내면 안 된다. 작업 순서 의존성을 명확히 해야 자동화가 거짓 성공을 만들지 않는다.
간단한 파이썬 스크립트에서도 이 구조를 만들 수 있다.
from datetime import date, datetime
from pathlib import Path
import json
import logging
import time
import uuid
LOG_DIR = Path('logs')
OUTPUT_DIR = Path('output')
LOG_DIR.mkdir(exist_ok=True)
OUTPUT_DIR.mkdir(exist_ok=True)
logging.basicConfig(
filename=LOG_DIR / 'daily_sales.log',
level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s'
)
class TransientError(Exception):
"""잠시 후 다시 시도하면 해결될 가능성이 있는 오류"""
class DataValidationError(Exception):
"""입력 데이터 수정 없이는 해결되지 않는 오류"""
def collect_data(context):
logging.info("collect_data start run_id=%s", context['run_id'])
# API 조회, DB 조회, 파일 다운로드 로직
# 일시적인 네트워크 오류라면 TransientError를 발생시킨다.
return {'raw_count': 120}
def validate_and_transform(context):
logging.info('validate_and_transform start')
raw = context['collect_data']
if raw['raw_count'] == 0:
raise DataValidationError('수집된 데이터가 없습니다.')
# 필수 열 검증, 날짜 변환, 금액 정리
return {'valid_count': 118, 'error_count': 2}
def create_report(context):
logging.info('create_report start')
output_path = OUTPUT_DIR / f"daily_sales_{context['work_date']}.xlsx"
# 보고서 파일 생성 로직
return {'output_path': str(output_path)}
def send_notification(context):
logging.info('send_notification start')
# Slack, Teams, 이메일 알림 발송
return {'notified': True}
def append_history(record):
with open(LOG_DIR / 'workflow_runs.jsonl', 'a', encoding='utf-8') as f:
f.write(json.dumps(record, ensure_ascii=False) + 'n')
def run_with_retry(
name,
func,
context,
retries=2,
delay_seconds=60,
retry_exceptions=(TransientError,)
):
for attempt in range(1, retries + 2):
try:
logging.info('task=%s attempt=%s', name, attempt)
result = func(context)
context[name] = result
logging.info('task=%s success', name)
return
except retry_exceptions:
context['failed_task'] = name
logging.exception('task=%s transient failure attempt=%s', name, attempt)
if attempt > retries:
raise
time.sleep(delay_seconds)
except Exception:
context['failed_task'] = name
logging.exception('task=%s failed without retry', name)
raise
def main():
context = {
'run_id': str(uuid.uuid4()),
'workflow_name': 'daily_sales_report',
'work_date': date.today().isoformat(),
'triggered_by': 'schedule',
'started_at': datetime.now().isoformat(),
'failed_task': None
}
status = 'success'
error_message = None
try:
run_with_retry('collect_data', collect_data, context)
run_with_retry('validate_and_transform', validate_and_transform, context, retries=0)
run_with_retry('create_report', create_report, context)
run_with_retry('send_notification', send_notification, context)
except Exception as e:
status = 'failed'
error_message = str(e)
logging.exception('workflow failed')
# 여기에서 장애 알림을 별도로 보낸다.
finally:
context['ended_at'] = datetime.now().isoformat()
context['status'] = status
context['error_message'] = error_message
append_history(context)
if __name__ == '__main__':
main()이 코드는 완전한 운영 시스템은 아니지만 중요한 뼈대를 담고 있다. 실행마다 run_id를 만들고, 단계별 로그를 남기며, 실패 시 실행 이력을 저장한다. 특히 TransientError처럼 일시적이라고 판단한 오류만 재시도하고, 데이터 검증 오류처럼 사람이 확인해야 하는 문제는 바로 멈추도록 구분했다. 나중에 Prefect나 Airflow로 옮기더라도 이 구조는 그대로 유지된다.
재시도는 모든 실패의 답이 아니다
재시도는 운영 자동화의 핵심 기능이다. 하지만 모든 오류를 무조건 다시 실행하면 안 된다.
재시도에 적합한 오류는 일시적인 오류다.
API 서버가 잠시 응답하지 않음
네트워크 연결이 끊김
호출 제한으로 몇 분 뒤 다시 시도해야 함
파일이 아직 업로드 중이라 잠시 기다려야 함
반대로 재시도해도 해결되지 않는 오류가 있다.
필수 열이 없는 엑셀 파일
잘못된 고객 코드
권한이 없는 API 토큰
보고서 수신자 목록 오류
승인되지 않은 지급 또는 발송 요청
이런 오류는 반복 실행하지 말고 멈춰야 한다. 그리고 담당자가 수정할 수 있도록 오류 파일, 로그 경로, 원본 데이터 위치를 알려야 한다.
재시도 설계에서는 다음을 정한다.
몇 번 재시도할 것인가
재시도 간격은 몇 초 또는 몇 분인가
어떤 오류만 재시도할 것인가
최종 실패 시 누구에게 알릴 것인가
실패한 작업만 다시 실행할 수 있는가
특히 외부 발송, 주문 상태 변경, 결제 요청처럼 업무 결과가 바뀌는 작업은 중복 실행에 주의해야 한다. 재시도 때문에 같은 이메일이 두 번 발송되거나 같은 주문이 두 번 처리되면 자동화가 오히려 사고를 만든다.
그래서 운영 자동화에서는 멱등성을 고려한다. 멱등성이란 같은 작업을 여러 번 실행해도 결과가 한 번 실행한 것과 같도록 만드는 성질이다. 예를 들어 보고서 발송 전에 notification_history 테이블을 확인해 이미 같은 날짜, 같은 보고서, 같은 수신자에게 보냈다면 다시 보내지 않는 방식이다.
실행 이력은 자동화의 블랙박스다
자동화가 실패했을 때 가장 먼저 묻는 질문은 정해져 있다.
언제 실행됐는가
어떤 입력 데이터를 사용했는가
어느 단계에서 실패했는가
몇 번 재시도했는가
결과 파일은 어디에 있는가
담당자에게 알림이 갔는가
이 질문에 답하려면 실행 이력을 남겨야 한다. 처음에는 workflow_runs.jsonl 같은 파일로 시작해도 된다. 업무가 커지면 데이터베이스 테이블로 옮기는 것이 좋다.
기본 실행 이력에는 다음 항목을 저장한다.
| 항목 | 설명 |
|---|---|
| run_id | 한 번의 실행을 구분하는 고유 ID |
| workflow_name | 작업 이름 |
| work_date | 실행 대상 날짜 또는 기간 |
| started_at, ended_at | 시작과 종료 시각 |
| status | success, failed, skipped 등 상태 |
| failed_task | 실패한 단계 이름 |
| error_message | 오류 요약 |
| input_count | 처리한 입력 건수 |
| output_path | 결과 파일 또는 저장 위치 |
| triggered_by | 스케줄 실행, 수동 실행, 이벤트 실행 여부 |
이력은 감사와 개선에도 필요하다. 매일 5분 걸리던 작업이 갑자기 40분 걸린다면 데이터가 늘었거나 외부 API가 느려졌을 수 있다. 특정 단계에서 실패가 반복된다면 코드를 고치는 것보다 입력 데이터 검증 규칙을 강화해야 할 수도 있다.
로그, 모니터링, 장애 알림 구성하기
로그는 개발자를 위한 기록이고, 알림은 담당자를 위한 행동 요청이다. 둘을 구분해야 한다.
로그에는 상세한 실행 정보를 남긴다.
작업 시작과 종료
입력 파일명과 처리 건수
API 상태 코드와 재시도 횟수
검증 실패 건수
결과 저장 위치
예외 메시지와 스택 트레이스
다만 개인정보, 비밀번호, API 토큰, 고객의 민감한 내용은 로그에 남기지 않는다. 로그는 문제 해결을 돕는 기록이지 민감 정보를 복제하는 장소가 아니다.
알림은 짧고 구체적이어야 한다. 실패 알림에는 다음 내용이 들어가면 좋다.
[자동화 실패] daily_sales_report
- 실행 ID: 4f3c...
- 실패 단계: validate_and_transform
- 대상 날짜: 2026-05-27
- 오류 요약: amount 열에 숫자로 변환할 수 없는 값 2건 발견
- 오류 파일: output/errors/daily_sales_2026-05-27.csv
- 다음 행동: 오류 파일 확인 후 원본 데이터를 수정하고 재실행성공 알림은 너무 자주 보내면 소음이 된다. 중요한 업무라면 성공 알림을 보내되 핵심 지표만 담는다. 일반적인 배치 작업은 실패 알림은 즉시 보내고, 성공 현황은 대시보드나 일일 요약으로 확인하는 방식이 낫다.
모니터링은 단순히 실패를 보는 것이 아니다. 다음 지표를 함께 본다.
최근 실행 성공률
평균 실행 시간
단계별 실패 횟수
재시도 발생 횟수
처리 건수의 급격한 증가 또는 감소
알림 발송 성공 여부
이 지표가 있어야 자동화가 조용히 망가지는 상황을 막을 수 있다. 예를 들어 작업은 성공으로 끝났지만 처리 건수가 평소 1,000건에서 0건으로 줄었다면 사실상 실패일 수 있다. 따라서 업무 기준의 이상 탐지도 필요하다.
Prefect와 Airflow로 확장하는 흐름
단순 스크립트가 커지면 워크플로 도구를 도입할 수 있다. Prefect에서는 앞의 함수를 작업 단위로 감싸고 흐름을 정의한다.
from prefect import flow, task
@task(retries=2, retry_delay_seconds=60)
def collect_data():
...
@task
def validate_and_transform(raw):
...
@task
def create_report(clean_data):
...
@task
def send_notification(report):
...
@flow
def daily_sales_report():
raw = collect_data()
clean_data = validate_and_transform(raw)
report = create_report(clean_data)
send_notification(report)이 구조의 장점은 작업별 상태가 분리된다는 점이다. 어느 단계가 실패했는지, 몇 번 재시도했는지, 다음 실행이 언제인지 확인하기 쉽다. 개발자는 여전히 파이썬 함수 중심으로 생각할 수 있다.
Airflow에서는 작업 간 의존성을 DAG로 표현한다. DAG는 방향이 있는 작업 그래프라고 이해하면 된다. 예를 들어 collect가 성공해야 transform이 실행되고, 그다음 report, notify가 실행되는 식이다. 여러 팀이 같은 플랫폼에서 수많은 배치 작업을 관리해야 한다면 이런 중앙 관리 방식이 유리하다.
중요한 것은 도구보다 설계다. cron으로 실행하든, Prefect를 쓰든, Airflow를 쓰든 다음 원칙은 같다.
작업을 작고 명확한 단위로 나눈다.
각 단계의 입력과 출력을 남긴다.
실패해도 원인을 추적할 수 있게 로그를 남긴다.
재시도 가능한 오류와 멈춰야 하는 오류를 구분한다.
중복 실행되어도 사고가 나지 않게 설계한다.
중요한 외부 발송이나 상태 변경에는 사람의 승인 절차를 둔다.
운영 전 점검할 것
자동화를 스케줄에 올리기 전에 다음 질문에 답해 보자.
이 작업은 어느 서버 또는 PC에서 실행되는가
실행 서버의 시간대와 환경 변수는 확인했는가
실행 계정에 필요한 최소 권한만 부여했는가
API 키와 비밀번호는 코드 밖에서 관리되는가
입력 데이터가 없을 때 정상 종료할 것인가, 실패로 볼 것인가
같은 날짜 작업이 두 번 실행되면 어떻게 되는가
실패 알림은 누구에게, 어떤 채널로 가는가
담당자가 알림을 받았을 때 무엇을 해야 하는지 알 수 있는가
로그와 실행 이력은 얼마 동안 보관하는가
수동 재실행 절차는 문서화되어 있는가
보고서 발송, 데이터 수정, 외부 요청에는 승인 단계가 필요한가
자동화 운영은 코드를 숨겨진 곳에서 돌리는 일이 아니다. 업무가 예측 가능하게 흐르도록 실행 조건, 기록, 알림, 재처리 절차를 만드는 일이다.
AI 자동화도 이 원칙 위에서 움직여야 한다. 문서 요약, 문의 분류, 보고서 초안 생성 같은 AI 작업은 비용과 응답 품질의 변동이 있다. 그래서 더더욱 실행 이력, 입력 데이터, 모델 호출 결과, 사람의 검토 여부를 남겨야 한다.
다음 장부터는 AI를 자동화 흐름에 넣는 방법을 다룬다. 하지만 기억해야 할 기준은 같다. 명확한 규칙으로 처리할 수 있는 일은 규칙 기반 자동화로 안정적으로 처리하고, 문맥 이해와 초안 작성처럼 규칙만으로 어려운 일에 AI를 붙인다. 그리고 그 AI 작업도 결국 스케줄, 로그, 재시도, 승인 절차 안에서 운영되어야 한다.
참고
이 노트는 보도·비평·교육·연구 목적으로 작성되었습니다. 저작권 문의가 있으시면 에서 알려주세요.