Перейти к содержанию

Интеграция: PostgreSQL (Mindbox) → API Gateway¶

Архитектура¶

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  PostgreSQL     │────▶│    Debezium     │────▶│   API Gateway   │
│  (БД Mindbox)   │     │  CDC Connector  │     │     (mTLS)      │
│                 │     │  + Avro Convert │     │                 │
└─────────────────┘     └─────────────────┘     └──────┬──────────┘
                                                        │
                                                        ▼
                                                 ┌─────────────────┐
                                                 │  Kafka raw      │
                                                 └─────────────────┘

Вариант 1: CDC через Debezium (Рекомендуется)¶

Типичные таблицы Mindbox в PostgreSQL¶

customers          - Клиенты
orders             - Заказы
order_items        - Товары в заказах
customer_actions   - События (просмотры, клики, etc)
product_catalog    - Каталог товаров

Конфигурация Debezium Connector¶

{
  "name": "mindbox-customers-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "mindbox-db.company.local",
    "database.port": "5432",
    "database.user": "debezium_user",
    "database.password": "${file:/secrets/debezium-password}",
    "database.dbname": "mindbox_prod",
    "database.server.name": "mindbox",

    "table.include.list": "public.customers,public.orders,public.order_items",

    "plugin.name": "pgoutput",
    "slot.name": "debezium_mindbox",
    "publication.name": "mindbox_publication",

    "transforms": "unwrap,addPrefix,route",

    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",

    "transforms.addPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.addPrefix.regex": "mindbox\\.public\\.(.*)",
    "transforms.addPrefix.replacement": "marketing.$1.raw"
  }
}

SQL для создания publication¶

-- На сервере PostgreSQL с БД Mindbox
CREATE PUBLICATION mindbox_publication FOR TABLE 
    public.customers,
    public.orders,
    public.order_items;

-- Проверка
SELECT * FROM pg_publication_tables WHERE pubname = 'mindbox_publication';

Вариант 2: Batch экспорт (Для исторических данных)¶

Python скрипт для batch загрузки¶

import psycopg2
import requests
import avro.io
import avro.schema
from io import BytesIO

# Подключение к БД Mindbox
conn = psycopg2.connect(
    host="mindbox-db.company.local",
    database="mindbox_prod",
    user="readonly_user",
    password="secure_password"
)

def extract_customers(since_date):
    """
    Извлечь клиентов из БД Mindbox (НЕ из API!)
    """
    cursor = conn.cursor()

    query = """
        SELECT 
            customer_id,
            email,
            phone,
            first_name,
            last_name,
            created_at,
            updated_at,
            segment_ids
        FROM customers
        WHERE updated_at >= %s
        ORDER BY updated_at ASC
    """

    cursor.execute(query, (since_date,))
    return cursor.fetchall()

def send_to_api_gateway(records, topic, version):
    """
    Отправить в API Gateway
    """
    # Сериализация в Avro
    avro_payload = serialize_to_avro(records, schema)

    response = requests.post(
        f"https://data-gateway.company.ru/api/1.0/{topic}",
        data=avro_payload,
        headers={
            "Content-Type": "application/avro",
            "version": version,
            "batch": str(len(records)),
        },
        cert=("marketing.customers.crt", "marketing.customers.key"),
        verify="ca.crt"
    )

    assert response.status_code == 201
    return response

if __name__ == "__main__":
    customers = extract_customers("2026-01-01")
    send_to_api_gateway(customers, "marketing.customers", "1.0.0")

Типичные данные из Mindbox¶

Customers (Клиенты)¶

SELECT 
    customer_id,
    email,
    phone,
    first_name,
    last_name,
    birthdate,
    city,
    total_orders_count,
    total_orders_amount,
    segment_ids,  -- JSON массив
    created_at,
    updated_at
FROM customers
WHERE is_active = true

Orders (Заказы)¶

SELECT 
    o.order_id,
    o.customer_id,
    o.order_number,
    o.total_amount,
    o.discount_amount,
    o.status,
    o.created_at,
    o.updated_at,
    array_agg(
        json_build_object(
            'product_id', oi.product_id,
            'quantity', oi.quantity,
            'price', oi.price
        )
    ) AS items
FROM orders o
JOIN order_items oi ON o.order_id = oi.order_id
WHERE o.updated_at >= NOW() - INTERVAL '1 day'
GROUP BY o.order_id

Рекомендации¶

Что использовать когда¶

Данные	Частота	Метод
Клиенты	Real-time	CDC (Debezium)
Заказы	Real-time	CDC (Debezium)
Действия клиентов	Real-time	CDC (Debezium)
Каталог товаров	1/день	Batch ETL
История (backfill)	Однократно	Batch ETL

Мониторинг¶

CDC lag - отставание Debezium
Replication slot size - не допускать переполнения
Schema changes - алерты при изменении структуры БД

Troubleshooting¶

Проблема: Нет доступа к БД Mindbox

Решение: Mindbox On-Premise - есть доступ к PostgreSQL. Mindbox Cloud - согласовать с вендором read-only доступ для CDC.

Проблема: Высокая нагрузка на БД

Решение: - Использовать Read Replica для CDC - Настроить rate limiting в Debezium - Использовать batch для неcritical данных

Дополнительные материалы¶