Что такое идемпотентность в контексте пайплайнов данных? Как вы обеспечиваете идемпотентность ваших процессов?

Идемпотентность в пайплайнах данных

Идемпотентность — это свойство операции, при котором многократное выполнение с одинаковыми входными данными дает одинаковый результат, без побочных эффектов.

Другими словами: запустил пайплайн один раз или 100 раз — результат в базе данных всё равно будет одинаковым.

Зачем нужна идемпотентность

В реальных системах:

• Пайплайны падают и перезагружаются
• Retry механизмы могут выполнить одну задачу дважды
• Network сбои приводят к дублированию запросов
• Scheduled jobs иногда запускаются несколько раз

Без идемпотентности результаты будут неправильными (дубли, неверные суммы, etc).

Примеры НЕ идемпотентных операций

-- ❌ НЕ идемпотентно: INSERT без проверки дубликатов
INSERT INTO sales (order_id, amount)
VALUES (123, 100);

-- Если выполнить дважды → две строки с order_id=123

-- ❌ НЕ идемпотентно: обновление без условия
UPDATE balance SET amount = amount + 100 WHERE user_id = 1;

-- Если выполнить дважды → amount вырастет на 200 вместо 100

-- ❌ НЕ идемпотентно: простое DELETE
DELETE FROM temp_table;

-- Если выполнить дважды → второй раз ничего не удалится, но это OK
-- Главное, что результат одинаковый

Примеры идемпотентных операций

-- ✅ Идемпотентно: INSERT OR REPLACE (UPSERT)
INSERT INTO sales (order_id, amount)
VALUES (123, 100)
ON CONFLICT (order_id) DO UPDATE SET amount = 100;

-- Выполни 100 раз → одна строка с order_id=123, amount=100

-- ✅ Идемпотентно: SET вместо UPDATE
UPDATE balance SET amount = 150 WHERE user_id = 1;

-- Выполни 100 раз → amount всегда 150

-- ✅ Идемпотентно: DELETE с условием (если ничего не найдёт, не ошибка)
DELETE FROM temp_table WHERE processing_date < '2024-01-01';

-- ✅ Идемпотентно: CREATE TABLE IF NOT EXISTS
CREATE TABLE IF NOT EXISTS users (id SERIAL, name VARCHAR);

-- ✅ Идемпотентно: SELECT (чтение не меняет состояние)
SELECT * FROM users;

Стратегии обеспечения идемпотентности

1. UPSERT (INSERT OR UPDATE/MERGE)

Вместо INSERT → UPSERT (вставить, или если существует → обновить)

-- PostgreSQL
INSERT INTO users (user_id, name, email, updated_at)
VALUES (1, 'Alice', 'alice@example.com', NOW())
ON CONFLICT (user_id) DO UPDATE SET
    name = EXCLUDED.name,
    email = EXCLUDED.email,
    updated_at = NOW();

-- MySQL
INSERT INTO users (user_id, name, email, updated_at)
VALUES (1, 'Alice', 'alice@example.com', NOW())
ON DUPLICATE KEY UPDATE
    name = VALUES(name),
    email = VALUES(email),
    updated_at = NOW();

-- SQL Server
MERGE INTO users AS target
USING (SELECT 1 as user_id, 'Alice' as name, 'alice@example.com' as email) as source
ON target.user_id = source.user_id
WHEN MATCHED THEN
    UPDATE SET name = source.name, email = source.email
WHEN NOT MATCHED THEN
    INSERT (user_id, name, email) VALUES (source.user_id, source.name, source.email);

2. Truncate + Full Reload (для некритичных таблиц)

-- Удалить всё за конкретный день и перезагрузить
DELETE FROM daily_stats WHERE date = '2024-03-20';
INSERT INTO daily_stats SELECT * FROM source WHERE date = '2024-03-20';

-- Выполни 100 раз → всегда одни данные за 2024-03-20

3. Change Data Capture (CDC) с уникальными ключами

# Отслеживаем только новые или изменённые записи
from debezium import PostgresConnector

connector = PostgresConnector()
changes = connector.get_changes(table='users')

for change in changes:
    # Каждое изменение имеет уникальный transaction_id
    if is_already_processed(change['transaction_id']):
        continue  # пропускаем дубликаты
    
    apply_change(change)
    mark_as_processed(change['transaction_id'])

4. Idempotency Keys (для асинхронных операций)

# При обработке события создаём уникальный ключ
import hashlib

def process_payment(event):
    # Создаём уникальный идентификатор операции
    idempotency_key = hashlib.md5(
        f"{event['user_id']}_{event['order_id']}_{event['timestamp']}".encode()
    ).hexdigest()
    
    # Проверяем, не обрабатывали ли этот ключ ранее
    if db.query("SELECT 1 FROM processed_operations WHERE idempotency_key = ?", idempotency_key):
        return  # уже обработана
    
    # Выполняем операцию
    process_payment_logic(event)
    
    # Записываем, что этот ключ обработан
    db.execute("INSERT INTO processed_operations (idempotency_key) VALUES (?)", idempotency_key)

5. Conditional Updates (UPDATE только при изменении)

-- UPDATE только если значение изменилось
UPDATE user_balance
SET balance = 1000, last_updated = NOW()
WHERE user_id = 1 AND balance != 1000;

-- Или со специальной колонкой версии
UPDATE user_balance
SET balance = 1000, version = version + 1, last_updated = NOW()
WHERE user_id = 1 AND version = 5;  -- обновляем только версию 5

Real-world пример: Idempotent ETL пайплайн

from datetime import datetime, timedelta
from sqlalchemy import create_engine, text
import pandas as pd

class IdempotentETL:
    def __init__(self):
        self.engine = create_engine("postgresql://user:pass@localhost/db")
    
    def load_daily_sales(self, date_str='2024-03-20'):
        """Загрузить продажи за день идемпотентно"""
        
        # 1. Удалить данные за эту дату (если есть)
        with self.engine.connect() as conn:
            conn.execute(
                text("DELETE FROM sales_daily WHERE date = :date"),
                {"date": date_str}
            )
            conn.commit()
        
        # 2. Загрузить свежие данные
        df = pd.read_sql(
            f"SELECT * FROM raw_sales WHERE date = '{date_str}'",
            self.engine
        )
        
        # 3. Трансформировать
        df['processed_at'] = datetime.now()
        df['amount'] = df['amount'].astype(float)
        df = df.dropna(subset=['order_id'])
        
        # 4. Загрузить в целевую таблицу
        df.to_sql('sales_daily', self.engine, if_exists='append', index=False)
        
        print(f"Loaded {len(df)} records for {date_str}")
    
    def upsert_customer_stats(self, date_str='2024-03-20'):
        """UPSERT статистику клиентов (идемпотентно)"""
        
        query = f"""
        -- Используем MERGE для идемпотентности
        MERGE INTO customer_stats cs
        USING (
            SELECT 
                customer_id,
                DATE(order_date) as stats_date,
                COUNT(*) as order_count,
                SUM(amount) as total_amount,
                NOW() as updated_at
            FROM raw_sales
            WHERE DATE(order_date) = '{date_str}'
            GROUP BY customer_id, DATE(order_date)
        ) src
        ON cs.customer_id = src.customer_id 
           AND cs.stats_date = src.stats_date
        WHEN MATCHED THEN
            UPDATE SET 
                order_count = src.order_count,
                total_amount = src.total_amount,
                updated_at = src.updated_at
        WHEN NOT MATCHED THEN
            INSERT (customer_id, stats_date, order_count, total_amount, updated_at)
            VALUES (src.customer_id, src.stats_date, src.order_count, src.total_amount, src.updated_at)
        """
        
        with self.engine.connect() as conn:
            conn.execute(text(query))
            conn.commit()
        
        print(f"Upserted customer stats for {date_str}")

# Использование
etl = IdempotentETL()

# Выполни это 100 раз — результат всегда будет одинаковым
for i in range(100):
    etl.load_daily_sales('2024-03-20')
    etl.upsert_customer_stats('2024-03-20')
    
# В базе данных будут данные только за 2024-03-20, без дубликатов

Best Practices

1. Вспомогательная таблица для отслеживания обработки

CREATE TABLE pipeline_runs (
    run_id UUID PRIMARY KEY,
    pipeline_name VARCHAR,
    date_partition DATE,
    status VARCHAR,  -- pending, running, success, failed
    started_at TIMESTAMP,
    ended_at TIMESTAMP
);

-- Перед запуском: INSERT с статусом 'running'
-- После успеха: UPDATE статус на 'success'
-- Если уже есть 'success' за эту дату → пропустить

2. Использовать date partition для reload

# Простой способ переделать данные за день
DATE_TO_PROCESS = '2024-03-20'
engine.execute(f"DELETE FROM data_table WHERE date = '{DATE_TO_PROCESS}'")
# ... загрузить заново

3. Логирование обработанных записей

CREATE TABLE processed_records (
    id BIGSERIAL PRIMARY KEY,
    source_id VARCHAR,
    process_type VARCHAR,
    idempotency_key VARCHAR UNIQUE,
    processed_at TIMESTAMP DEFAULT NOW()
);

-- Проверяем идемпотентность:
SELECT COUNT(*) FROM processed_records 
WHERE idempotency_key = 'unique_key';

Частые ошибки

-- ❌ Плохо: UPDATE без проверки
UPDATE balance SET amount = amount + 100 WHERE user_id = 1;
-- Если выполнить дважды → неправильный результат

-- ✅ Хорошо: SET вместо += 
UPDATE balance SET amount = 200 WHERE user_id = 1;
-- Выполни 100 раз → результат всегда 200

-- ❌ Плохо: простой INSERT без проверки
INSERT INTO users VALUES (1, 'Alice');
-- Может привести к ошибке PRIMARY KEY

-- ✅ Хорошо: INSERT OR IGNORE
INSERT INTO users VALUES (1, 'Alice')
ON CONFLICT DO NOTHING;

Заключение

Идемпотентность — это не опция, а требование для надёжных пайплайнов. Всегда проектируй операции так, чтобы они безопасно повторялись несколько раз.