Эффективные массовые вставки с ON CONFLICT COPY для обрабо... | Вопросы для собеседования

Эффективные массовые вставки с ON CONFLICT COPY для обработки данных высокого объема

Вопрос:

У нас есть таблица user со следующими столбцами: user_id, email, last_login. Поля user_id и email должны быть уникальными. Мы хотим записать набор записей для обновления last_login. Некоторые записи из нашего набора имеют user_id как в таблице (известные пользователи). Некоторые пользователи новые. Таким образом, нам потребуется операция "upsert".

Можете описать, как разрешить конфликты по уникальному полю email при обновлении отметки времени последнего входа?

Подсказки:

Рассмотрите использование ключевого слова ON CONFLICT.
Стратегии пакетной обработки могут значительно улучшить производительность по сравнению с операциями по строкам.
Команда COPY может быть полезна для начальной загрузки данных, но требует дополнительных шагов по обработке конфликтов.
Подумайте об управлении транзакциями и влиянии ограничений (constraints) на производительность.

Выше ожиданий:

Знание CTID и стратегий временных таблиц для оптимизированных обновлений.
Понимание расширений pg_bulkload или аналогичных.
Реализация касмотным разрешений конфликтов с функциями на стороне сервера БД.

Ответ:

Использование ON CONFLICT (UPSERT)

Самый прямой подход к массовым обновлениям (bulk upsert) — использование встроенного синтаксиса INSERT ON CONFLICT PostgreSQL:

INSERT INTO users (user_id, email, last_login)
VALUES 
    (1, 'user1@example.com', NOW()),
    (2, 'user2@example.com', NOW()),
    (3, 'user3@example.com', NOW())
ON CONFLICT (email) 
DO UPDATE SET last_login = EXCLUDED.last_login;

Это выполняет атомарную операцию, которая вставляет новые строки или обновляет существующие, когда возникают конфликты по полю email.

Обработка данных партиями для повышения производительности

Для больших наборов данных необходимо обрабатывать данные порциями, а не сразу:

batch_size = 5000
total_records = len(data)

for i in range(0, total_records, batch_size):
    batch = data[i:i+batch_size]
    
    # Construct VALUES part of query
    values = []
    for record in batch:
        values.append(f"({record.id}, '{record.email}', '{record.last_login}')")
    
    query = f"""
    INSERT INTO users (user_id, email, last_login)
    VALUES {','.join(values)}
    ON CONFLICT (email) DO UPDATE 
    SET last_login = EXCLUDED.last_login;
    """
    
    execute_query(query)

Команда COPY с использованием временной таблицы

Для чрезвычайно больших наборов данных используйте команду COPY для более быстрой начальной загрузки:

-- Создание временной таблицы с такой же структурой
CREATE TEMP TABLE temp_users (LIKE users INCLUDING DEFAULTS);

-- Использование COPY для массового вставки во временную таблицу (в коде приложения)
COPY temp_users (user_id, email, last_login) FROM 'path/to/data.csv' WITH CSV HEADER;

-- Выполнение upsert из временной таблицы в основную таблицу
INSERT INTO users
SELECT * FROM temp_users
ON CONFLICT (email) DO UPDATE
SET last_login = EXCLUDED.last_login;

-- Очистка
DROP TABLE temp_users;

Продвинутый метод, основанный на CTID для максимальной производительности

Этот скрипт демонстрирует эффективный шаблон для обработки массовых обновлений данных в PostgreSQL, используя временные таблицы и команду COPY для минимизации накладных расходов транзакций.

CTID — это системный столбец в PostgreSQL, который уникально идентифицирует физическое расположение строки в таблице. Он представляет собой комбинацию номера блока и индекса кортежа в этом блоке.

-- Создание временной таблицы с данными
CREATE TEMP TABLE temp_users AS 
SELECT * FROM users WITH NO DATA;

-- Загрузка данных во временную таблицу с помощью COPY
COPY temp_users FROM 'data.csv' WITH CSV HEADER;

-- Добавление столбца флага "обработан"
ALTER TABLE temp_users ADD COLUMN processed boolean DEFAULT false;

-- Выполнение обновлений на основе сканирования CTID
WITH updated AS (
    UPDATE users u
    SET last_login = t.last_login
    FROM temp_users t
    WHERE u.email = t.email
    RETURNING t.ctid
)
UPDATE temp_users
SET processed = true
WHERE ctid IN (SELECT ctid FROM updated);

-- Вставка оставшихся записей
INSERT INTO users (user_id, email, last_login)
SELECT user_id, email, last_login 
FROM temp_users
WHERE NOT processed;

Управление транзакциями

Оборачивайте операции в транзакцию, чтобы обеспечить атомарность:

BEGIN;

-- Ваши операции upsert здесь

COMMIT;

Для очень больших операций рассмотрите:

Временное отключение триггеров
Увеличение параметра work_mem
Установка synchronous_commit = off (с осторожностью)

Настройка разрешения конфликтов с помощью функций

Для сложной логики разрешения конфликтов:

CREATE OR REPLACE FUNCTION resolve_user_conflict() 
RETURNS TRIGGER AS $$
BEGIN
    IF NEW.last_login > OLD.last_login THEN
        -- Обновлять только если новый вход более поздний
        OLD.last_login := NEW.last_login;
    END IF;
    RETURN OLD;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER user_conflict_trigger
BEFORE UPDATE ON users
FOR EACH ROW EXECUTE FUNCTION resolve_user_conflict();

Затем используйте с обычными операциями upsert.

SQL Средний Опубликовано