Как вы бы спроектировали надёжный ETL-конвейер? Какие основные принципы?

Проектирование надёжного ETL-конвейера

Основные принципы

ETL (Extract-Transform-Load) — процесс извлечения данных из источников, трансформации и загрузки в хранилище. Надёжность — критический аспект ETL.

1. Идемпотентность

Главный принцип: один запуск = несколько запусков с одинаковым результатом.

Структура ETL должна быть безопасной при повторном выполнении:

# Плохо: не идемпотентно
def load_data(source_df, target_table):
    target_table.insert(source_df)  # Дублирует при повторе

# Хорошо: идемпотентно
def load_data(source_df, target_table):
    # Сначала очистить целевую таблицу за день
    delete_from_target(date=TODAY)
    # Потом загрузить свежие данные
    target_table.insert(source_df)

Использование Upsert (Update-Insert)

-- PostgreSQL: ON CONFLICT для идемпотентности
INSERT INTO customers (id, name, email, updated_at)
VALUES (1, 'John', 'john@example.com', NOW())
ON CONFLICT (id) DO UPDATE SET
    name = EXCLUDED.name,
    email = EXCLUDED.email,
    updated_at = NOW();

2. Обработка ошибок и восстановление

Retry логика с экспоненциальной задержкой

import time
from typing import Callable, Any

def retry_with_backoff(
    func: Callable,
    max_attempts: int = 3,
    base_delay: int = 2
) -> Any:
    """Повтор с экспоненциальной задержкой"""
    for attempt in range(1, max_attempts + 1):
        try:
            return func()
        except Exception as e:
            if attempt == max_attempts:
                raise
            delay = base_delay ** attempt
            print(f"Ошибка: {e}. Повтор через {delay}сек...")
            time.sleep(delay)

# Использование
data = retry_with_backoff(
    lambda: fetch_data_from_api(),
    max_attempts=3
)

Transactional Load

BEGIN TRANSACTION;

-- Этап 1: валидация
CREATE TEMP TABLE staging_data AS
SELECT * FROM import_data
WHERE is_valid = true;

-- Этап 2: трансформация
CREATE TEMP TABLE transformed_data AS
SELECT 
    TRIM(name) as name,
    LOWER(email) as email,
    CAST(age AS INTEGER) as age
FROM staging_data;

-- Этап 3: загрузка
DELETE FROM target_table WHERE date = TODAY;
INSERT INTO target_table SELECT * FROM transformed_data;

-- Либо всё успешно, либо откат
COMMIT;

3. Валидация данных

Data Quality Checks

class DataValidator:
    def __init__(self, df):
        self.df = df
        self.errors = []
    
    def validate_schema(self, required_columns):
        """Проверить наличие колонок"""
        missing = set(required_columns) - set(self.df.columns)
        if missing:
            self.errors.append(f"Отсутствуют колонки: {missing}")
        return self
    
    def validate_not_null(self, columns):
        """Проверить отсутствие NULL"""
        for col in columns:
            null_count = self.df[col].isnull().sum()
            if null_count > 0:
                self.errors.append(f"Null в {col}: {null_count} строк")
        return self
    
    def validate_range(self, column, min_val, max_val):
        """Проверить диапазон значений"""
        invalid = self.df[(self.df[column] < min_val) | (self.df[column] > max_val)]
        if not invalid.empty:
            self.errors.append(f"{column} вне диапазона [{min_val}, {max_val}]")
        return self
    
    def validate_unique(self, columns):
        """Проверить уникальность"""
        duplicates = self.df[columns].duplicated().sum()
        if duplicates > 0:
            self.errors.append(f"Дубликаты в {columns}: {duplicates}")
        return self
    
    def is_valid(self):
        return len(self.errors) == 0
    
    def raise_if_invalid(self):
        if not self.is_valid():
            raise ValueError("\n".join(self.errors))

# Использование
validator = DataValidator(df)
validator.validate_schema(['id', 'name', 'email'])\
         .validate_not_null(['id'])\
         .validate_range('age', 0, 150)\
         .validate_unique(['email'])\
         .raise_if_invalid()

4. Логирование и мониторинг

import logging
from datetime import datetime

logger = logging.getLogger(__name__)

class ETLPipeline:
    def __init__(self, name: str):
        self.name = name
        self.start_time = None
        self.metrics = {}
    
    def run(self):
        self.start_time = datetime.now()
        logger.info(f"ETL '{self.name}' started")
        
        try:
            self.extract()
            self.transform()
            self.load()
            
            duration = (datetime.now() - self.start_time).total_seconds()
            logger.info(f"ETL '{self.name}' completed in {duration}s")
            logger.info(f"Metrics: {self.metrics}")
            
        except Exception as e:
            logger.error(f"ETL '{self.name}' failed: {e}", exc_info=True)
            # Отправить алерт
            self.send_alert(f"ETL failed: {e}")
            raise
    
    def extract(self):
        logger.info("Extract phase started")
        rows = self.fetch_source()
        self.metrics['extracted_rows'] = len(rows)
        logger.info(f"Extracted {len(rows)} rows")
        return rows
    
    def transform(self):
        logger.info("Transform phase started")
        # Трансформация
        self.metrics['transformed_rows'] = 1000
    
    def load(self):
        logger.info("Load phase started")
        # Загрузка
        self.metrics['loaded_rows'] = 1000
        logger.info(f"Loaded {self.metrics['loaded_rows']} rows")
    
    def send_alert(self, message: str):
        # Отправить в Slack, email, Datadog, etc
        pass

5. Управление зависимостями

DAG-ориентированный подход (Directed Acyclic Graph)

# Использование Apache Airflow или Dagster
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'start_date': datetime(2024, 1, 1),
}

with DAG(
    'etl_pipeline',
    default_args=default_args,
    schedule_interval='@daily',
) as dag:
    
    # Этап 1: загрузка из API
    extract_task = PythonOperator(
        task_id='extract_from_api',
        python_callable=extract_data
    )
    
    # Этап 2: валидация
    validate_task = PythonOperator(
        task_id='validate_data',
        python_callable=validate_data
    )
    
    # Этап 3: трансформация
    transform_task = PythonOperator(
        task_id='transform_data',
        python_callable=transform_data
    )
    
    # Этап 4: загрузка в БД
    load_task = PythonOperator(
        task_id='load_to_warehouse',
        python_callable=load_data
    )
    
    # Определить порядок выполнения
    extract_task >> validate_task >> transform_task >> load_task

6. Управление состоянием и метаданными

-- Таблица для отслеживания запусков ETL
CREATE TABLE etl_runs (
    run_id UUID PRIMARY KEY,
    pipeline_name VARCHAR(100),
    status VARCHAR(20),  -- pending, running, success, failed
    started_at TIMESTAMPTZ,
    completed_at TIMESTAMPTZ,
    rows_processed INTEGER,
    rows_failed INTEGER,
    error_message TEXT
);

-- Таблица для отслеживания изменений (Change Data Capture)
CREATE TABLE cdc_log (
    change_id BIGSERIAL PRIMARY KEY,
    table_name VARCHAR(100),
    operation VARCHAR(10),  -- INSERT, UPDATE, DELETE
    record_id UUID,
    old_data JSONB,
    new_data JSONB,
    changed_at TIMESTAMPTZ
);

7. Масштабируемость

Partitioning и параллелизм

from concurrent.futures import ThreadPoolExecutor
from typing import List

def parallel_extract(sources: List[str], workers: int = 4):
    """Параллельная загрузка из разных источников"""
    with ThreadPoolExecutor(max_workers=workers) as executor:
        results = list(executor.map(fetch_from_source, sources))
    return results

# Партиционирование по дате
def load_partitioned(df, table_name: str, date_column: str):
    """Загрузить данные отдельно по датам"""
    for date, group in df.groupby(pd.Timestamp(date_column).dt.date):
        partition = f"{table_name}_{date.strftime('%Y%m%d')}"
        group.to_sql(partition, engine)

8. Чеклист надёжного ETL

• ✓ Идемпотентность (безопасность при повторах)
• ✓ Валидация данных на каждом этапе
• ✓ Graceful error handling и retry логика
• ✓ Transactional load (всё или ничего)
• ✓ Полное логирование и мониторинг
• ✓ Управление зависимостями (DAG)
• ✓ Ведение метаданных (audit trail)
• ✓ Тестирование (unit, integration, data quality)
• ✓ Документация и процедуры восстановления
• ✓ Мониторинг SLA и оповещения

Надёжный ETL-конвейер — это инвестиция в стабильность и доверие к данным в организации.