데이터 파이프라인 모니터링

Python으로 ETL과 ELT

Jake Roach

Data Engineer

데이터 파이프라인 모니터링

데이터 파이프라인은 데이터 변화와 실행 실패를 모니터링해야 합니다

누락 데이터
데이터 타입 변동
패키지 폐기 또는 기능 변경

데이터 파이프라인의 로깅 워크플로

파이프라인 성능 로깅

실행 시 성능을 기록합니다
실패 시 원인 파악의 출발점을 제공합니다

import logging
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.DEBUG)

# Create different types of logs
logging.debug(f"Variable has value {path}")
logging.info("Data has been transformed and will now be loaded.")

DEBUG: Variable has value raw_file.csv
INFO: Data has been transformed and will now be loaded.

경고와 오류 로깅

import logging
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.DEBUG)

# Create different types of logs
logging.warning("Unexpected number of rows detected.")
logging.error("{ke} arose in execution.")

WARNING: Unexpected number of rows detected.
ERROR: KeyError arose in execution.

try-except으로 예외 처리

try:
    # Execute some code here
    ...

except:
    # Logging about failures that occured
    # Logic to execute upon exception
    ...

오류 발생 시 실행할 코드를 정의합니다

try-except으로 특정 예외 처리

except 절에 구체적 예외를 전달하세요

try:
    # Try to filter by price_change
    clean_stock_data = transform(raw_stock_data)
    logging.info("Successfully filtered DataFrame by 'price_change'")

except KeyError as ke:
    # Handle the error, create new column, transform
    logging.warning(f"{ke}: Cannot filter DataFrame by 'price_change'")
    raw_stock_data["price_change"] = raw_stock_data["close"] - raw_stock_data["open"]
    clean_stock_data = transform(raw_stock_data)

Passons à la pratique !

Python으로 ETL과 ELT