การตรวจสอบ data pipeline

ETL and ELT ใน Python

Jake Roach

Data Engineer

การตรวจสอบ data pipeline

ควรตรวจสอบ data pipeline เพื่อดูการเปลี่ยนแปลงของข้อมูลและความล้มเหลวในการทำงาน

ข้อมูลขาดหาย
ชนิดข้อมูลเปลี่ยนแปลง
Package เลิกใช้งานหรือฟังก์ชันการทำงานเปลี่ยนแปลง

เวิร์กโฟลว์การบันทึกล็อกสำหรับ data pipeline

การบันทึกล็อกประสิทธิภาพของ data pipeline

บันทึกประสิทธิภาพการทำงานขณะรัน
เป็นจุดเริ่มต้นเมื่อโซลูชันล้มเหลว

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 clause

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)

มาฝึกกันเถอะ!

ETL and ELT ใน Python