🌤️ 공휴일/주말 만족도 분석 파이프라인 구축기 (feat. 기상청 API)

Jinyoung Cheon·2025년 6월 12일

1. 프로젝트 개요

주제: 기상청 Open API를 활용해 다양한 주제의 데이터 파이프라인을 팀별로 구성
팀 전략: 하나의 큰 주제를 '기상청 API 활용'으로 잡고, 각자 다른 세부 주제로 파이프라인을 설계
내 파트: "공휴일 및 주말의 날씨 만족도 분석"
기간: 2025.06.03 ~ 2025.06.12
기술 스택:
- requests, holidays, datetime, pandas
- Apache Airflow
- Snowflake
- Slack Webhook

2. 내가 맡은 역할

기상청 날씨 정보 API에서 텍스트 포맷 데이터를 수집하고 파싱하여 사용
holidays 패키지를 사용해 공휴일 여부 판별, 주말 포함 여부 판단
날짜별로 기온, 강수량, 상대습도 등을 활용하여 만족도 점수 산정
Airflow DAG로 전체 흐름을 자동화하고, Snowflake에 최종 결과 테이블 적재
Slack Webhook을 이용한 결과 알림 자동화까지 포함한 End-to-End 설계

3. DAG 구성

[기상청 날씨 API]
	↓ extract
[필요한 지역/기간 데이터 수집]
	↓ transform
[주말/공휴일 필터링 → 요약 → 점수 산출]
	↓ load
[Snowflake 또는 내부 DB에 적재]
	↓ notify
[Slack 메시지로 알림 전송]

extract(region_name: str) (TASK)
- 기능: 지역별 주간 날씨 데이터를 기상청 API에서 수집
- 로직:
  - 월요일에 DAG 실행 시, 전 주 월요일~일요일 데이터 요청
  - region_name → region_code로 변환 후 요청
  - 요청 기간: YYYYMMDD0000 ~ YYYYMMDD2300
  - 결과: 날씨 데이터 리스트 (records) 반환

transform(extracted: dict) (TASK)
- 기능: 주말 또는 공휴일 데이터 필터링 후, 날씨 요약 및 야외 점수 계산
- 로직:
  - 날짜 기준으로 is_weekend 또는 is_holiday 조건 필터링
  - summarize_weather()로 평균 기온, 습도, 풍속, 강수량 요약
  - calculate_outdoor_score()로 점수 산출
load(transformed: dict) (TASK)
- 기능: 계산된 점수를 DB 또는 Snowflake에 저장
- 로직:
  - insert_weather_score() 함수 호출
  - 각 지역별 요약 정보 + 점수 → 테이블에 저장
  - None이면 적재 스킵
notify(transformed_listL List[dict]) (TASK)
- 기능: 상위/하위 지역 3곳씩 선정 후, Slack 메시지 전송
- 로직:
  - 점수 기준으로 내림차순 정렬
  - 상위 3개, 하위 3개 지역 각각 메시지 포맷팅
  - Slack 메시지 전송 (내용: 지역명, 점수, 날씨 요약, 날짜 등)

4. 기술적 도전 & 해결 과정

🔎 문제 1: Backfill 명령어 실행했는데 데이터가 안 들어감

Airflow DAG 구성:

with DAG(
    dag_id="weather_pipeline_dag",
    start_date=datetime(2025, 1, 6),  # 월요일 시작
    schedule_interval="@weekly",
    catchup=True,
    ...
)

백필 명령 실행:

airflow dags backfill -s 2025-01-02 -e 2025-06-02 weather_pipeline_dag

문제: backfill을 실행했음에도 불구하고 과거 데이터가 DB에 저장되지 않았다 😱

✅ 해결

- backfill은 실행 이력이 없는 날짜에만 작동  
- DAG 실행 이력이 있거나 실패 기록이 있으면 해당 날짜는 건너뜀  
- 해결: Airflow UI에서 DAG를 삭제하고 다시 등록한 후 backfill 실행

📚 배운 점

실행 이력이 있는 날짜는 backfill 대상이 아님
개발 중엔 DAG 초기화가 빠른 해결책일 수 있음

🔎 문제 2: 지역별 데이터를 처리하기 위한 동적 태스크 매핑

전국 데이터를 한 태스크에서 처리하면 코드 복잡성과 실패 위험 증가

✅ 해결

extracted_list = extract.expand(region_name=regions)
transformed_list = transform.expand(extracted=extracted_list)
loaded_list = load.expand(transformed=transformed_list)
notify.override(trigger_rule=TriggerRule.ALL_SUCCESS)(loaded_list)

📚 배운 점

expand()로 리스트 기반 병렬 처리 가능
병렬 처리 + 개별 실패 감지에 효과적

🔎 문제 3: 데이터 멱등성(idempotency) 처리

DAG 재실행 시 중복 적재 방지 필요
MERGE 문으로 동일 날짜의 데이터를 업데이트하거나 새로 삽입

✅ 해결

MERGE INTO WEATHER_DB.JINYOUNG.WEATHER_SCORE tgt
USING (
    SELECT '{region_code}' AS region_code, '{date}' AS date, ...
) src
ON tgt.region_code = src.region_code AND tgt.date = src.date
WHEN MATCHED THEN
    UPDATE SET ...
WHEN NOT MATCHED THEN
    INSERT (...) VALUES (...);

📚 배운 점

MERGE 문은 멱등성과 최신성 동시 보장
Airflow 재실행/백필에 매우 유용한 방식

🔎 문제 4: Task 실행 오류 디버깅의 연속

NoneType, 리스트 인덱스 에러, Slack 메시지 포맷 오류 등 다양한 문제가 발생

✅ 해결

로그 확인 → 변수 출력 → 조건 분기 처리 순서로 디버깅

if not filtered:
    print("주말/공휴일 데이터 없음, 적재 스킵")
    return None

📚 배운 점

로그 분석과 변수 추적이 에러 해결의 핵심
예외 상황을 고려한 방어적 코딩 중요

🔎 문제 5: Slack 봇 연동 및 알림 포맷 설계

모든 지역 데이터를 수집하고 만족도 점수를 계산한 뒤, 결과를 팀 Slack 채널에 자동 알림으로 전송
상위 3개 지역과 하위 3개 지역을 만족도 기준으로 정렬하여 시각적으로 보기 좋게 포맷팅

✅ 해결

def send_slack_message(message: str):
    webhook_url = Variable.get("slack_webhook_url")  
    payload = {"text": message}
    response = requests.post(webhook_url, json=payload)

    if response.status_code == 200:
        print("✅ Slack 전송 성공")
    else:
        print(f"❌ Slack 전송 실패: {response.status_code}, {response.text}")