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

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)

연습해 봅시다!

Python으로 하는 ETL과 ELT