Почему стоит использовать инкрементальную загрузку данных

Для решения проблем хранения и обработки больших данных возникает потребность в репликации данных из классического хранилища‑источника в аналитическое, чтобы делать аналитику без влияния на продуктивную эксплуатацию. К тому же скорость работы аналитических запросов в таких базах в разы выше, чем в традиционных.

При изменении операционных данных источника, аналитическое хранилище становится неактуальным. Для обеспечения актуальности данных, аналитическому хранилищу необходимо эти данные обновлять, минимизируя задержку в доставке данных и нагрузку на систему-источник.

Наивный подход, заключающийся в простой перезагрузке данных в аналитическое хранилище – неэффективен, потому что как правило, только небольшая часть исходных данных изменяется во время циклов загрузки. А современные исследования лишь подтверждают данный факт (например, работа J¨org, T., Dessloch, S.: “Formalizing ETL Jobs for Incremental Loading of Data Warehouses”). Поэтому желательно фиксировать эти изменения в оперативных источниках данных и обновлять в аналитическом хранилище только те данные, которые изменились. Данный подход известен как инкрементная загрузка или захват изменений.

Для решения данной проблемы используются процессы ETL (Extract, Transform, Loading) с использованием паттерна "Change Data Capture" (CDC), которые актуализируют хранилище посредством периодического обновления данных, которые были изменены.

Современные методы реализации Change Data Capture на источнике

Внедрения паттерна Change Data Capture в процесс репликации данных включает в себя следующие этапы:

Сбор измененных данных на источнике
Доставка измененных данных
Имплементация измененных данных в аналитическое хранилище

Каждый из этапов может быть оптимизирован, но в данной статье рассматривается только этап оптимизации на классическом реляционном источнике, в качестве которого выступает БД Oracle.

При анализе существующих научных работ по данной теме, можно выделить четыре базовые методологии реализации паттерна Change Data Capture на источнике, наиболее точная классификация которых представлена ниже:

TABLE COMPARISON, который подразумевает полное сравнение старой копии источника и актуальной его версии
TIMESTAMPS – временная метка, представляющая собой столбец, в который записывается дата и время изменения записи в базе данных
DATABASE TRIGGERS – метод, основанный на применении DML – триггера, который исполняется каждый раз, когда пользователь пытается изменить данные
LOG READERS ON DATABASES использует системные журналы транзакций, которые базы данных генерируют для собственного восстановления в случае сбоя, и извлекает из них данные, которые были изменены

От части уже было проведено сравнение этих подходов. Например, в исследовании H. Chandra "Analysis of Change Data Capture Method in Heterogeneous Data Sources to Support RTDW" отслеживали время работы данных методов на популярных реляционных БД-источниках, в числе которых был и Oracle, где лучше всего себя показал метод DATABASE TRIGGERS. Но в тоже время в работе Z. Bin, S. Shuai, G. Zhi-chun, and H. Jian-feng, ‘‘Design and implementation of incremental data capturing in wireless network planning based on log mining’’ по результатам исследования, уже подход LOG READERS ON DATABASES называют самым эффективным и прогрессивным для БД Oracle. Данные результаты вызывают необходимость в более тщательном тестировании, чтобы подтвердить или опровергнуть выше полученные результаты, какой из методов показывает наилучшую пропускную способность, и действительно ли методы TABLE COMPARISON и TIMESTAMPS не нужны в современных реалиях.

Сравнение методов Change Data Capture на источнике

В данной же статье, предлагается сравнить предложенные выше CDC-подходы для репликации данных из целевого хранилища в аналитическое в разрезе их оптимизации на системе источника, которым будет выступать СУБД Oracle, по заранее выбранным критериям.

Для системы источника важно постоянно находится в работоспособном состоянии, поскольку с ней чаще всего связано и функционирование других систем, поэтому излишняя нагрузка на базу данных без крайней необходимости – непозволительна. Таким образом, в первую очередь необходимо сравнить данные подходы по их влиянию на производительность системы источника, по следующим выбранным критериям:

Elapsed Time(mins) – время работы метода, от начала изменения данных до полной их репликации в целевое хранилище
% of total CPU for Instance – процент потребления процессора
Total (MB) Read + Write per second – общий объём чтения и записи данных на диск за секунду времени
max % Host Mem used for SGA(system global area) + PGA(program global area) - процент максимального потребления оперативной памяти в ходе выполнения. Oracle использует два типа структур памяти – общую и относящуюся к процессу. Системная глобальная область (system global area – SGA) – это часть общей памяти, которую разделяют между собой все серверные процесс (включая фоновые). Специфичная для процессов часть памяти известна как программная глобальная область (program global area - PGA)

Для достижения поставленной цели будут рассмотрены реализации всех четырех CDC-подходов, обеспечивающих получение изменённых данных для их последующей репликации в заданное целевое хранилище. В качестве тестового стенда был выбран следующий набор технологий. База данных-источник базировалась на Oracle DB, которая функционировала на личной машине. Слой CDC был реализован с помощью языка PL/SQL.

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

Процессор: AMD Ryzen 5 5500U (6 ядер/12 потоков, 2,1-4,0 ГГц, L3-кэш 8 МБ, TDP 10-25 Вт), техпроцесс 7 нм
Оперативная память: 16 ГБ, 3200 МГц, DDR4
Накопитель: 512 ГБ NVMe PCIe SSD SSSTC CL1-8D512
Видеокарта: AMD Radeon Graphics

Полное сравнение таблиц (TABLE COMPARISON)

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

--вставленные и изменённые строки
SELECT * FROM new_version
MINUS
SELECT * FROM old_version;
  
--удалённые строки
SELECT * FROM old_version
MINUS
SELECT * FROM new_version;

Тогда в общем случае мы имеем следующим алгоритм получения изменений:

Получить текущий снимок базы данных
Получить вставленные и изменённые строки как разность текущего и прошлого снимков
Получить удаленные строки как разность прошлого и текущего снимков
Передать полученные строки в интерфейс
Сохранить текущий снимок как прошлый

Предложенный механизм получения изменений обладает двумя достоинствами: простотой и универсальностью. В общем случае запрос, формирующий текущий снимок, может соединять много исходных таблиц и содержать различные бизнес-условия. Исходный запрос — это единственная уникальная часть, которую нужно разработать с нуля при реализации данного механизма. Все шаги алгоритма остаются без изменения при любом базовом запросе с любым количеством исходных таблиц. Чем сложнее базовый запрос, чем больше таблиц и условий он вовлекает, тем более выгодно предложенное решение по сравнению с накопителем дельты на триггерах, журнальных таблицах и процедуре, реализующей уникальную логику.

Минусом предложенного решения является необходимость хранить последний снимок, который занимает место в БД. Если речь идет о дополнительном хранении от сотен до сотен тысяч строк, вряд ли это может представлять проблему для современных СУБД. Дисковое пространство обходится дешевле разработки и поддержки изощренных решений, специальных для каждого набора исходных таблиц и столбцов. Если снимок содержит миллионы строк или требует много времени для формирования, тогда, вероятно, стоит обратиться к другому решению. Также дифференцирование таблиц не может захватить данные, которые вернулись к своим старым значениям. Например, если значение столбца менялось несколько раз между старой и новой версиями таблицы.

Для тестовой реализации рассмотрим транзакционную таблицу ACC_TRANSACTION, которая содержит историю операций некоторой платежной системы. С учетом предложенного выше алгоритм, для реализации необходимо создать ещё вспомогательные и интерфейсную таблицы.

Сначала подготовим таблицы:

-- исходная таблица
create table ACC_TRANSACTION (
    TXN_ID number(19,0) not null,
    AMOUNT float not null,
    FUNDS_AVAIL_DATE timestamp not null,
    TXN_DATE timestamp not null,
    TXN_TYPE_CD varchar2(10 char),
    ACCOUNT_ID number(10,0),
    primary key (TXN_ID)
);

-- интерфейсная таблица
create table ACC_TRANSACTION_TABLE_COMPARISON (
    TXN_ID number(19,0),
    AMOUNT float,
    FUNDS_AVAIL_DATE timestamp,
    TXN_DATE timestamp,
    TXN_TYPE_CD varchar2(10 char),
    ACCOUNT_ID number(10,0),
    deleted number(1) default 0 not null -- 1 deleted 0 alive
);

-- прошлый снимок
create table ACC_TRANSACTION_ (
    TXN_ID number(19,0) not null,
    AMOUNT float not null,
    FUNDS_AVAIL_DATE timestamp not null,
    TXN_DATE timestamp not null,
    TXN_TYPE_CD varchar2(10 char),
    ACCOUNT_ID number(10,0),
    primary key (TXN_ID)
);

-- текущий снимок и дельта
create table ACC_TRANSACTION_tt_ (
    TXN_ID number(19,0) not null,
    AMOUNT float not null,
    FUNDS_AVAIL_DATE timestamp not null,
    TXN_DATE timestamp not null,
    TXN_TYPE_CD varchar2(10 char),
    ACCOUNT_ID number(10,0),
    delta number(1) default 0 /*не дельта*/ not null,
    primary key (TXN_ID)
);

Далее создадим процедуру, реализующую данный алгоритм:

create or replace procedure process_ACC_TRANSACTION_delta
is
begin
    -- 1. получить текущий снимок базовым запросом
    insert into ACC_TRANSACTION_tt_ (
        TXN_ID,
        AMOUNT,
        FUNDS_AVAIL_DATE,
        TXN_DATE,
        TXN_TYPE_CD,
        ACCOUNT_ID,
        delta
    )
    select
        TXN_ID,
        AMOUNT,
        FUNDS_AVAIL_DATE,
        TXN_DATE,
        TXN_TYPE_CD,
        ACCOUNT_ID,
        0 /*не дельта*/
    from ACC_TRANSACTION
    ;
    -- 2. получить дельту-плюс как разность текущего и прошлого снимков
    merge into ACC_TRANSACTION_tt_ tgt
    using ( -- строки, которых не было в прошлый раз
        select
            TXN_ID,
            AMOUNT,
            FUNDS_AVAIL_DATE,
            TXN_DATE,
            TXN_TYPE_CD,
            ACCOUNT_ID
        from ACC_TRANSACTION_tt_
        minus
        select
            TXN_ID,
            AMOUNT,
            FUNDS_AVAIL_DATE,
            TXN_DATE,
            TXN_TYPE_CD,
            ACCOUNT_ID
        from ACC_TRANSACTION_
    ) src
    on (src.txn_id = tgt.txn_id)
    when matched then
        update set delta = 1 /*дельта-плюс*/
    ;
    -- 3. получить дельту-минус
    merge into ACC_TRANSACTION_tt_ tgt
    using ( -- строки, которые были в прошлый раз
        select
            TXN_ID,
            AMOUNT,
            FUNDS_AVAIL_DATE,
            TXN_DATE,
            TXN_TYPE_CD,
            ACCOUNT_ID 
        from ACC_TRANSACTION_
    ) src
    on (src.txn_id = tgt.txn_id)
    when not matched then
        insert (
            TXN_ID,
            AMOUNT,
            FUNDS_AVAIL_DATE,
            TXN_DATE,
            TXN_TYPE_CD,
            ACCOUNT_ID,
            delta
        ) values (
            src.TXN_ID,
            src.AMOUNT,
            src.FUNDS_AVAIL_DATE,
            src.TXN_DATE,
            src.TXN_TYPE_CD,
            src.ACCOUNT_ID,
            2/*дельта-минус*/
        )
    ;
    -- 4. передать дельту-плюс и -минус в интерфейс
    insert into ACC_TRANSACTION_TABLE_COMPARISON (
        TXN_ID,
        AMOUNT,
        FUNDS_AVAIL_DATE,
        TXN_DATE,
        TXN_TYPE_CD,
        ACCOUNT_ID,
        deleted
    ) select
        TXN_ID,
        AMOUNT,
        FUNDS_AVAIL_DATE,
        TXN_DATE,
        TXN_TYPE_CD,
        ACCOUNT_ID,
        case delta when 2 then 1 else 0 end
    from ACC_TRANSACTION_tt_
    where delta != 0 /*дельта-плюс и минус*/
    ;
    -- 5. сохранить текущий снимок как прошлый
    execute immediate 'truncate table ACC_TRANSACTION_';
    insert into ACC_TRANSACTION_ (
        TXN_ID,
        AMOUNT,
        FUNDS_AVAIL_DATE,
        TXN_DATE,
        TXN_TYPE_CD,
        ACCOUNT_ID  
    ) select
        TXN_ID,
        AMOUNT,
        FUNDS_AVAIL_DATE,
        TXN_DATE,
        TXN_TYPE_CD,
        ACCOUNT_ID
    from ACC_TRANSACTION_tt_
    where delta != 2 /*кроме дельты-минус*/
    ;
    execute immediate 'truncate table ACC_TRANSACTION_tt_';
end process_ACC_TRANSACTION_delta;
/

Для того чтобы поставить данную процедуру на регулярное самостоятельное периодическое выполнение, можно воспользоваться процедурой из системного пакетом планирования активностей в СУБД Oracle:

-- Запускаем job на постоянное обновление интерфейсной таблицы
begin
dbms_scheduler.create_job(
         job_name        => 'CDC_TABLE_COMARISON',
         job_type        => 'STORED_PROCEDURE',
         job_action      => 'process_ACC_TRANSACTION_delta',
         start_date      => systimestamp,
         end_date        => null,
         repeat_interval => 'freq = secondly',
         enabled         => true,
         auto_drop       => false,
         comments        => 'Change Data Capture based on table comparison');
end;

После выполнения данного скрипта ручных действий для запуска процедуры больше не требуется, она будет автоматически выполняться с частотой раз в секунду. Частота обновления интерфейсной таблицы будет зависеть от параметра repeat_interval, который в качестве аргумента принимает конфигурационную строку. Остальные параметры обновления проставлены в соответствии с официальной документацией.

Временная метка (TIMESTAMPS)

Таблицы, изменения в которых должны быть зафиксированы, могут иметь столбец, представляющий время последнего изменения. Если исходная система более сложная, то на строке также может быть установлено состояние "флаг" или номер версии. В таких случаях может быть выполнен запрос, чтобы извлечь только те строки, которые были изменены после (или между) определенного параметра даты и времени:

--timestamp, status indicators and versioning on rows
SELECT * 
  FROM source_table
 WHERE 1 = 1
   AND date_time_stamp > TO_TIMESTAMP('2023/03/24 10:13:18', 'YYYY/MM/DD HH:MI:SS')
   AND status_flag = 'U'
   AND version_code > 2;

Однако у этого метода также есть несколько ограничений:

Накладные расходы на захват измененных данных должны покрываться исходной базой данных, так как придется выполнять потенциально дорогостоящие запросы к таблицам исходной базы. Необходимость в этих запросах может заставить администратора БД добавить индексы для оптимизации, которые в противном случае были бы не нужны. Также нет возможности перенести эту нагрузку на staging-базу данных.

Этот метод не решает проблему метода TABLE COMPARISON в захвате промежуточных значений. Если значение столбца в таблице колеблется, то не получиться захватить все промежуточные значения или даже узнать, изменялся ли столбец, если конечное значение такое же, как и в последний раз, когда были зафиксированы изменения данных.

Для того чтобы на постоянной основе передавать изменения в интерфейс, реализуем следующую тестовую процедуру, которая будет передавать полученные результаты запроса в соответствующую интерфейсную таблицу:

create or replace procedure get_ACC_TRANSACTION_timestamp
is
begin
    insert into ACC_TRANSACTION_TIMESTAMP (
        TXN_ID,
        AMOUNT,
        FUNDS_AVAIL_DATE,
        TXN_DATE,
        TXN_TYPE_CD,
        ACCOUNT_ID
    )
    select
        TXN_ID,
        AMOUNT,
        FUNDS_AVAIL_DATE,
        TXN_DATE,
        TXN_TYPE_CD,
        ACCOUNT_ID
   from ACC_TRANSACTION
  where TXN_DATE >= sysdate - interval '1' second;
end get_ACC_TRANSACTION_timestamp;
/

И точно также поставим процедуру на ежесекундное обновление, что будет соответствовать временному интервалу, с которым мы отбираем записи в исходной таблице:

-- Запускаем job на постоянное обновление интерфейсной таблицы
begin
dbms_scheduler.create_job(
         job_name        => 'CDC_TIMESTAMP',
         job_type        => 'STORED_PROCEDURE',
         job_action      => 'get_ACC_TRANSACTION_timestamp',
         start_date      => systimestamp,
         end_date        => null,
         repeat_interval => 'freq = secondly',
         enabled         => true,
         auto_drop       => false,
         comments        => 'Change Data Capture based on timestamp');
end;

Триггеры базы данных (DATABASE TRIGGERS)

Немного более сложным методом для отслеживания изменений в базе данных является использование триггеров базы данных. Триггер — это особая разновидность хранимой процедуры, которая автоматически выполняется при возникновении события на сервере базы данных. Триггеры DML выполняются, когда пользователь пытается изменить данные с помощью событий языка обработки данных (DML). Событиями DML являются процедуры INSERT, UPDATE или DELETE, применяемые к таблице или представлению. Метод триггеров заключается в создании триггера DML-операций для исходных таблиц. Когда таблица изменяется, триггер вступает в силу и сохраняет инкрементальные данные в таблице идентификации инкрементальных данных. Обычно таблица содержит информацию о первичном ключе и идентификатор операции. Далее инкрементальные данные уже можно обрабатывать при помощи ETL-инструментов.

Как уже было сказано ранее, метод, основанный на триггерах, требует сохранения захваченных инкрементальных данных в таблице идентификации. Для этого в тестовой базе данных создадим таблицу идентификации ACC_TRANSACTION_TRIGGERS, которая будет хранить имена изменённых таблиц, первичные ключи и значения первичных ключей, а также типы DML операций и время срабатывания триггера. Первичным ключом в данном случае будет выступать столбец RECORD_ID, последовательно нумерующий строки.

create table ACC_TRANSACTION_TRIGGERS(
    RECORD_ID NUMBER, --record number
    T_NAME VARCHAR2(20), --source table name
    T_KEY VARCHAR2(20), --source table primary key name
    T_KEYVALUE1 VARCHAR2(20), --primary key value of new table
    T_KEYVALUE2 VARCHAR2(20), --primary key value of original table
    D_TYPE VARCHAR2(5), --DML operation type
    D_TIME VARCHAR2(100)  --DML operation time
);

Также необходимо создать триггер на исходную таблицу источника для каждой CRUD операции:

--создание триггера на операцию вставки
CREATE OR REPLACE TRIGGER TRIGGER_I
AFTER INSERT ON ACC_TRANSACTION
FOR EACH ROW BEGIN
INSERT INTO ACC_TRANSACTION_TRIGGERS(RECORD_ID, T_NAME, T_KEY, T_KEYVALUE1, D_TYPE,D_TIME)
VALUES(D_ID.NEXTVAL, 'ACC_TRANSACTION', 'TXN_ID', :NEW.TXN_ID, 'I', systimestamp);
END; 

--создание триггера на операцию удаления
CREATE OR REPLACE TRIGGER TRIGGER_D
AFTER DELETE ON ACC_TRANSACTION
FOR EACH ROW BEGIN
INSERT INTO ACC_TRANSACTION_TRIGGERS(RECORD_ID, T_NAME, T_KEY, T_KEYVALUE2, D_TYPE, D_TIME)
VALUES(D_ID.NEXTVAL, 'ACC_TRANSACTION', 'TXN_ID', :OLD.TXN_ID, 'D', systimestamp);
END;

--создание триггера на операцию обновления
CREATE OR REPLACE TRIGGER TRIGGER_U 
BEFORE UPDATE OF Txn_Id, Amount, Funds_Avail_Date, Txn_Date, Txn_Type_CD, Account_Id ON ACC_TRANSACTION
FOR EACH ROW BEGIN
INSERT INTO ACC_TRANSACTION_TRIGGERS(RECORD_ID, T_NAME, T_KEY, T_KEYVALUE1, T_KEYVALUE2, D_TYPE, D_TIME)
VALUES(D_ID.NEXTVAL, 'ACC_TRANSACTION', 'TXN_ID', :NEW.TXN_ID,  
   :OLD.TXN_ID, 'U', systimestamp);
END;

Несмотря на то, что метод триггеров всё равно создает нагрузку на исходную базу данных во время захвата, особенно если отслеживаются изменения в большом количестве таблиц, в сравнении же с предыдущими методами TABLE COMPARISON и TIMESTAMPS, этот подход может снизить затраты на извлечение измененных данных. Также стоит отметить, что этот метод является специфичным для базы данных, поэтому его обновление и миграция могут быть трудоёмкими.

Чтение журнала транзакций (LOG READERS ON DATABASES)

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

Потенциальные проблемы данного подхода:

Стоимость: в отличие от других вариантов, может возникнуть потребность в покупке дополнительных компонентов для чтения и обработки этого журнала
Согласование чтения журналов транзакций и архивирования файлов журналов. В некоторых случаях использование архивных журналов может быть выгодно, потому что есть возможность обработки на отдельной системе, что еще больше снизит нагрузку на источник, но вместе с тем это понесёт дополнительные сложности настройки
Работа с изменениями формата журналов транзакций между версиями системы управления базами данных
Устранение незафиксированных изменений, которые база данных записала в журнал транзакций и позже откатила
Работа с изменениями метаданных таблиц в базе данных

Я пометил вышеуказанные проблемы как потенциальные, потому что большинство программ для чтения журналов CDC справляются с этими проблемами «из коробки». Например, инструменты CDC, такие как GoldenGate и Attunity, позволяют обрабатывать только зафиксированные изменения. У них есть процессы для автоматизации обновления структур метаданных, и они должны автоматически идентифицировать версию базы данных, чтобы изменения в структуре журнала транзакций учитывались при выпусках.

С точки зрения преимуществ, данная технология имеет ряд определенных плюсов по сравнению с ранее обсуждаемыми вариантами, в том числе:

Минимальное воздействие на базу данных - тем более, если использовать пересылку журналов для обработки журналов на выделенном хосте.
Нет необходимости программных изменений в приложениях, которые используют базу данных
Низкая задержка при получении изменений. Эти технологии могут справляться с тысячами изменений в секунду
Транзакционная целостность: сканирование журнала может производить поток изменений, который повторяет исходные транзакции в порядке их фиксации. Такие потоки изменений включают изменения, внесенные во все таблицы, участвующие в захваченной транзакции

Хотя подробные операции с данными записываются в журнале Oracle, формат данных закрыт для доступа, их нельзя прочитать напрямую, для этого необходимы соответствующие инструменты, например Log Miner. Данный инструмент анализа журналов предоставляется Oracle 8i и более поздними версиями и может быть использован для получения операторов DML операций базы данных для резервного копирования, восстановления и дампа. Несмотря на существование технологии Golden Gate, еще одно коммерческое программное обеспечение от Oracle, которое на самом деле является более мощным и дорогостоящим инструментом, имеет ряд ограничений для разработки специализированной системы. Использование встроенного Log Miner от Oracle может удовлетворить большей части требований гибкой настройки анализа журнала для Oracle database и СDC.

Oracle redo log является гарантией целостности и непротиворечивости базы данных. В нем записывается вся информация о модификации базы данных, включая идентификатор транзакции, тип операции, объект операции, старое значение данных и новое значение данных. В процессе CDC рассматриваются только зафиксированные транзакции с данными. Анализируя записи операций insert, delete и update в redo log, можно получать инкрементные данные. Для этого сначала анализируются архивные журналы базы данных на основе Log Miner и создается представление содержимого управления журналами V$LOGMNR_CONTENTS. Затем необходимо сформировать запрос с учётом условий, в соответствии с которыми вы хотите отобрать инкрементальные данные. После этого необходимо устранить избыточность DML-операций и собрать данные в единый словарь для их последующей репликации.

Схема получения изменённых данных с помощью журнала транзакций выглядит так:

И включает она следующие шаги:

Перевести базу данных в режим инкрементирования журнала

ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

Добавить log-файл, статус группы которого CURRENT

--смотрим статистику по имеющимся группам log-файлов
SELECT group#, sequence#, bytes, archived, members, status 
  FROM V$LOG 
 WHERE status = 'CURRENT';

--ищем надейной группе соответствующий log-файл
SELECT group#, type, member 
  FROM V$LOGFILE;

--добавляем найденный log-файл
begin
    	dbms_logmnr.add_logfile('/путь к файлу/', dbms_logmnr.new);
end;

Запустить LogMiner

begin
    	dbms_logmnr.start_logmnr(
        	options => dbms_logmnr.DICT_FROM_ONLINE_CATALOG);
end;

Создать процедуру сбора изменённых данных из журнала в интерфейсную таблицу

create or replace procedure get_ACC_TRANSACTION_logbased
is
begin
    insert into ACC_TRANSACTION_LOG_BASED (
        REC_ID,
        T_NAME,
        D_TYPE,
        USER_NAME,
        D_TIME,
        SQL_REDO
    )
    select
        RECORD_ID.NEXTVAL,
        TABLE_NAME,
        OPERATION,
        USERNAME,
        TIMESTAMP,
        SQL_REDO
   from v$logmnr_contents
  where TIMESTAMP >= systimestamp - interval '1' second
    and TABLE_NAME = 'ACC_TRANSACTION';
end get_ACC_TRANSACTION_logbased;
   /

Поставить процедуру на ежесекундное выполнение, что будет соответствовать интервалу, с которым процедура собирает данные

-- Запускаем job на постоянное обновление интерфейсной таблицы
begin
dbms_scheduler.create_job(
         job_name        => 'CDC_LOG_BASED',
         job_type        => 'STORED_PROCEDURE',
         job_action      => 'get_ACC_TRANSACTION_logbased',
         start_date      => systimestamp,
         end_date        => null,
         repeat_interval => 'freq = secondly',
         enabled         => true,
         auto_drop       => false,
         comments        => 'Change Data Capture based on logs');
end;

Профильное нагрузочное тестирование

Для численной оценки производительности архитектур, было проведено профильное нагрузочное тестирование, при котором генерировались CRUD операции для тестовой OLTP-таблицы и производилась фиксация изменённых данных в соответствующей интерфейсной таблице. Инструментом для сбора статистики выступали AWR-отчёты, которые позволяют получить широкий спектр показателей производительности базы данных, но в качестве основных были выбраны следующие показатели:

Elapsed Time(mins)
% of total CPU for Instance
Total (MB) Read + Write per second
max % Host Mem used for SGA(system global area) + PGA(program global area)

Сама статистика рассматривается в разрезе snapshot’ов базы данных, а именно для генерации конкретно AWR-отчёта необходимо указать snapshot, с которого мы начинаем отслеживать статистику, и snapshot, после которого мы заканчиваем отслеживать статистику.

--создание snapshot'a базы данных
dbms_workload_repository.create_snapshot;

--просмотр статистики по snapshot'ам
select snap_id, dbid, instance_number, begin_interval_time, end_interval_time
  from dba_hist_snapshot 
order by begin_interval_time desc;

--запрос для просмотра AWR-отчёта в текстовом формате
select * from table(dbms_workload_repository.awr_report_text(
                                                l_dbid     => ..., 
                                                l_inst_num => ..., 
                                                l_bid      => ..., 
                                                l_eid      => ...));

Для получения более полных результатов было проведено тестирование для следующих объемов: 1000, 10000, 100000, 1000000 записей. Также особое внимание уделялось выбору профилей для нагрузочного тестирования и были выбраны следующие метрики для их формирования:

Распределение по типу операции - процентное соотношение между генерируемыми CRUD операциями: вставкой, обновлением, удалением
Распределение данных внутри таблицы. Для тестовой OLTP-таблицы можно применить условное секционирование. Если почти все операции применяются к одной секции, то такая секция называется горячей. Для нашей тестовой OLTP-таблицы можно привести аналогию с реальным кейсом, когда один покупатель совершает несколько транзакций подряд, тогда все CRUD операции будут применены лишь к одной секции, связанной с этим покупателем. Тогда наличие или отсутствие горячей секции будет одной из метрик при профильном тестирование

На основании описанных метрик были сформированы следующие нагрузочные профили:

Профиль	Распределение операций	Горячая секция
Профиль 1	40% вставка 40% обновление 20% удаление	Нет
Профиль 2	100% вставка	Нет
Профиль 3	10% вставка 90% обновление	Нет
Профиль 4	10% вставка 90% обновление	Да

Полученные результаты

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

Как видно из полученных результатов, практически для всех профилей методы DATABASE TRIGGERS и LOG READERS ON DATABASES показывают меньшую пропускную способность в сравнении с остальными подходами. Лучше всего себя показал метод TIMESTAMPS, что можно объяснить отличной работой запросов по индексированным высокоселективным столбцам в транзакционных базах данных, но в тоже время необходимость в наличии такого столбца в базе данных накладывает свои потребности на поддержку как самого столбца, так и существующих по нему индексов.

Также рассмотрим графики выборочных средних остальных отслеживаемых метрик, которые дают информацию о нагрузке на железо. Выборки также состояли из 10 запусков:

По нагрузке на процессор и диски все методы выдали сравнимый перфоманс, за исключением метода DATABASE TRIGGERS, который показывает значительно меньшую нагрузку на диски в сравнении с другими методами.

По использованию ресурсов оперативной памяти – можно сложить однозначный вывод, поскольку все тестируемые профили выдали схожие тенденции:

Однозначно можно сказать, что самым оптимальным методом по использованию ресурсов оперативной памяти является DATABASE TRIGGERS, а самым требовательным LOG READERS ON DATABASES. Самые простые методы в лице TABLE COMPARISON и TIMESTAMPS опять же выдали средние, схожие друг с другом показатели.

Таким образом, обобщив результаты всех профилей, получим следующую таблицу оптимальных методов по каждому из критерию:

Критерий	Оптимальный метод
Elapsed Time(mins)	TIMESTAMP
% of total CPU for Instance	TRIGGERS
Total (MB) Read + Write per second	TRIGGERS
max % Host Mem used for SGA+PGA	TRIGGERS

Анализ результатов

Как видно из полученных результатов, практически для всех профилей тестирования метод DATABASE TRIGGERS показывает значительно худшее время работы в сравнении с остальными, но в тоже время показывает наилучшие показатели по нагрузке на процессор, память и диски.

Для методов TABLE COMPARISON и TIMESTAMPS прослеживаются схожие тенденции в показателях во всех профилях, но нельзя забывать проблемы данных подходов, а именно эти методы фиксируют не все типы изменений. С одной стороны у TABLE COMPARISON дела с этим обстоят немного лучше, а с другой TIMESTAMPS может идти «из коробки», просто следуя из архитектуры реплицируемого хранилища.

По методу LOG READERS ON DATABASES можно отметить перекос в сторону ресурсов памяти, но в остальным выдал средние показатели схожие с предыдущими двумя методами. Также нельзя не отметить важное преимущество в переносе части этой нагрузки на staging-базу данных. Архивируя журналы на источнике и читая их на staging-базе данных, можно значительно снизить нагрузку. Однако данное усовершенствование ещё сильнее повысит сложность внедрения и поддержки и без того сложного подхода.

Итоги

По итогам работы можно сделать вывод, что все подходы имеют сравнимые друг с другом показатели, которые всё-таки стоит рассматривать в разрезе тех нюансов реализации, которые были описаны для каждого метода, тем самым опровергая бесполезность самых простых методов, для некоторого слоя задач они будут самым оптимальным решением.

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

Оптимизация Change Data Capture в БД Oracle