Инкрементальная загрузка данных

Об инкрементальной загрузке данных

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

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

Инкрементальная загрузка данных работает при использовании нескольких нод ClickHouse. Однако, существуют некоторые особенности:

В файле /var/lib/visiology/scripts/v3/custom-configs/cluster-formula-engine.json списки узлов EtlNodes и QueryNodes должны быть идентичными. Например:
"EtlNodes": ["clickhouse1", "clickhouse2", "clickhouse3"], "QueryNodes": ["clickhouse1", "clickhouse2", "clickhouse3"]
Это необходимо для корректной работы инкрементальной загрузки и выполнения запросов.
Параметры TargetReplicaNodesCount и ConfirmationReplicaNodesCount должны иметь одинаковое значение, равное общему количеству узлов в кластере.
Это гарантирует, что данные будут храниться на всех нодах, обеспечивая отказоустойчивость и согласованность. Например:
"TargetReplicaNodesCount": 3, "ConfirmationReplicaNodesCount": 3

Настройка инкрементальной загрузки данных

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

Откройте набор данных, в который вы хотите добавить таблицу с инкрементальной загрузкой, нажмите Данные в верхней панели и щелкните Получить данные в выпадающем меню:
В открывшемся окне выберите База данных и нажмите Далее:
Создайте новое подключение к серверу или выберите существующее в в левой части окна:
В правой части окна активируйте тумблер Инкрементальное обновление и заполните следующие поля:
- Название таблицы – название таблицы которое будет отображаться в модели данных.
- Текст запроса – текст запроса для получения данных. Запрос будет выполнять инкрементальную загрузку данных за определенный период времени, указанный обязательными параметрами @{RangeStart} и @{RangeEnd}. Данные параметры рассчитываются автоматически на основе значений, которые вы введёте в полях Начало архивации данных и Начало инкрементального обновления. Началом инкрементального периода будет окончание архивного периода. Пример запроса (PostgreSQL):
  select brandname, colorname, classname, productkey, size, stylename, manufacturer from dimproduct where date_key>=date(@{RangeStart}) and date_key < date (@{RangeEnd});
  где
  - SELECT brandname, colorname, classname, productkey, size, stylename, manufacturer – столбцы, которые выбираются из таблицы dimproduct.
  - WHERE date_key >= date(@{RangeStart}) AND date_key < date(@{RangeEnd}) – условие фильтрации данных по дате. date_key – название столбца, содержащего даты (это может быть любое название, которое вы назначили столбцу). Данный столбец должен обязательно присутствовать в таблице. Параметры @{RangeStart} и @{RangeEnd} определяют начальное и конечное значение диапазона дат соответственно. Дата преобразуется к типу DATE с помощью функции date(). Таким образом, запрос выбирает все записи, у которых дата (date_key) находится между значениями @{RangeStart} и @{RangeEnd}. Это означает, что будут выбраны только те строки, которые были изменены или добавлены в указанный временной интервал.
    В зависимости от используемой СУБД, вы можете использовать другие функции вместо date() в вашем SQL-запросе.
- Архивировать данные старше – определяет момент времени, начиная с которого данные будут архивироваться при загрузке. Данные, попадающие в этот временной интервал, загружаются только один раз. Если данные внутри этого диапазона изменятся в источнике, они повторно загружаться не будут. Архивный период будет автоматически увеличиваться с течением времени по мере поступления новых данных.
- Перезагружать данные за последние – определяет точку отсчёта, с которой начинается инкрементальная загрузка данных. Например, если указано значение “2 дня до даты обновления”, то обновятся данные, которые появились в течении двух дней до начала обновления. Это значит, что система будет собирать и обрабатывать изменения, произошедшие за эти два дня, чтобы включить их в итоговую версию данных. Важно отметить, что эти «два дня» могут меняться в зависимости от текущей даты. С каждым новым днём он сдвигается вперёд, охватывая новые данные. Этот механизм называют “скользящим окном". При этом архивный период будет увеличиваться, как описано выше.
  Пример:
  Допустим, мы впервые обновили данные 30 января 2025 г. с теми настройками, как показано на картинке выше. Загрузились данные за последние 7 дней до 20.01.2025. Из них архивным периодом станет 5 дней, а инкрементальным – 2 дня (включая текущий).
  На следующий день мы обновим данные. В этом случае загрузятся данные, которые изменились в источнике за 2 дня до нашего обновления, а архивный период увеличится на 1 день.
  Через 2 дня аналогично – данные за последние 2 дня обновятся, а архивный период увеличится.
  Если мы обновим таблицу через неделю, снова загрузятся только те изменения, которые были внесены в источник за последние два дня, а архивный период будет сдвинут на неделю вперед.
  Вы можете задавать разные комбинации для архивного и инкрементного периодов. Например, архивный период может составлять 3 месяца, а инкрементальный – 1 день.
Нажмите Добавить, чтобы добавить таблицу в набор данных. Теперь вы можете связать ее с таблицей фактов, а при необходимости её можно будет обновить вручную или автоматически. Таблица обновится согласно заданным настройкам инкрементальной загрузки.

Внимание!

Платформа выполняет загрузку последовательными временными окнами таким образом, что следующий параметр {@DateStart} всегда равен предыдущему {@DateEnd}. Поэтому запрос должен задавать период, исключая правую границу (значение конца периода), иначе одно и то же значение на стыке периодов попадёт в два окна и появятся дубли.

Во избежание дублирования используйте исключающую правую границу:

SELECT *
FROM orders
WHERE order_timestamp >= {@DateStart}
AND order_timestamp <  {@DateEnd}

Следующие ситуации приводят к дублям:

Включающая правая граница

... WHERE order_timestamp >= {@DateStart}

     AND order_timestamp <= {@DateEnd}

Использование BETWEEN в запросе инкрементальной загрузки:

... WHERE order_timestamp BETWEEN {@DateStart} AND {@DateEnd}

BETWEEN включает обе границы (эквивалент >= и <=), поэтому строки на точном значении {@DateEnd} загрузятся дважды.

Откройте набор данных, в который вы хотите добавить таблицу с инкрементальной загрузкой, нажмите Данные в верхней панели и щелкните Получить данные в выпадающем меню:
В открывшемся окне выберите требуемую базу данных из списка и нажмите Далее. Мы выберем PostgreSQL:
Создайте новое подключение к серверу или выберите существующее в в левой части окна:
В правой части окна выберите таблицы для загрузки и нажмите Далее. Например:
В открывшемся окне выберите таблицу, для которой вы хотите настроить инкрементальное обновление. В нашем примере это factonlinesales:
В правой части окна активируйте тумблер Инкрементальное обновление и заполните поля. Например:
- Источник диапазона – столбец таблицы, который содержит дату и служит маркером «новизны» строки. На основе значений в этом столбце система определяет, какие строки считать новыми (добавленными после предыдущей загрузки), а какие — подлежащими архивации. Пример выбора: datekey.
- Начинать обновление, если дата – задаёт оператор сравнения для начала диапазона загружаемых данных. Обычно используется «Больше или равно».
- Заканчивать обновление, если дата – задаёт оператор сравнения для конца диапазона. Обычно используется «Меньше».
- Архивировать данные старше – определяет момент времени, начиная с которого данные будут архивироваться при загрузке. Данные, попадающие в этот временной интервал, загружаются только один раз. Если данные внутри этого диапазона изменятся в источнике, они повторно загружаться не будут. Архивный период будет автоматически увеличиваться с течением времени по мере поступления новых данных.
- Перезагружать данные за последние – определяет точку отсчёта, с которой начинается инкрементальная загрузка данных. Например, если указано значение “2 дня до даты обновления”, то обновятся данные, которые появились в течении двух дней до начала обновления. Это значит, что система будет собирать и обрабатывать изменения, произошедшие за эти два дня, чтобы включить их в итоговую версию данных. Важно отметить, что эти «два дня» могут меняться в зависимости от текущей даты. С каждым новым днём он сдвигается вперёд, охватывая новые данные. Этот механизм называют “скользящим окном". При этом архивный период будет увеличиваться, как описано выше.
  Пример:
  Допустим, мы впервые обновили данные 30 января 2025 г. с теми настройками, как показано на картинке выше. Загрузились данные за последние 7 дней до 20.01.2025. Из них архивным периодом станет 5 дней, а инкрементальным – 2 дня (включая текущий).
  На следующий день мы обновим данные. В этом случае загрузятся данные, которые изменились в источнике за 2 дня до нашего обновления, а архивный период увеличится на 1 день.
  Через 2 дня аналогично – данные за последние 2 дня обновятся, а архивный период увеличится.
  Если мы обновим таблицу через неделю, снова загрузятся только те изменения, которые были внесены в источник за последние два дня, а архивный период будет сдвинут на неделю вперед.
  Вы можете задавать разные комбинации для архивного и инкрементного периодов. Например, архивный период может составлять 3 месяца, а инкрементальный – 1 день.
Нажмите Добавить, чтобы добавить таблицы в набор данных. Теперь вы можете связать их, а при необходимости их можно будет обновить вручную или автоматически. Таблица, для которой настроена инкрементальная загрузка обновится согласно заданным настройкам.

Если в источнике изменится структура таблицы (например, если добавятся новые столбцы или у столбцов изменится тип данных), то инкрементальная загрузка перестанет работать. В этом случае вам придётся заново загрузить таблицу и настроить для нее инкрементальную загрузку.
Числовые значения, которые вы указываете для параметров Архивировать данные старше и Перезагружать данные за последние не должны превышать 120 (как для месяцев, так и для дней).

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

Внутренняя логика работы инкрементальной загрузки

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

Разбиение данных на партиции:
- Данные разбиваются на партиции на основе указанной в настройках гранулярности (год, месяц, день). Это позволяет управлять загрузкой данных и минимизировать объем работы.
- Разбиение данных на партиции помогает избежать лишней обработки и ускоряет процесс инкрементальной загрузки.
Генерация запросов к источнику данных:
- При инкрементальной загрузке генерируется несколько запросов к источнику. Каждый запрос отвечает за загрузку данных для конкретной партиции.
- Количество запросов зависит от выбранной гранулярности и количества периодов, на которые делятся данные.
- Запросы на загрузку выполняются параллельно.
Управление уже загруженными данными:
- Уже загруженные данные не удаляются, благодаря использованию функционала ClickHouse для работы с партициями (например, с помощью команды ATTACH PARTITION). Это позволяет обновлять только новые или измененные данные, сохраняя уже загруженную информацию без изменений.
Работа с ключом партиции:
- Для каждой таблицы в системе добавляется служебная колонка partitionKey, которая используется для управления партициями в процессе загрузки.
- Эта колонка обеспечивает корректную идентификацию и обработку данных, а также позволяет эффективно работать с партициями.

ПРИМЕР

Предположим, что вы указали 3 месяца для архивного периода и 3 месяца для инкрементального периода, а текущей датой является 21.01.2025:

В этом случае система рассчитает даты следующим образом:

Начало архивного периода:
21.01.2025 - 3 мес = 01.10.2024 (дата “округляется“ до первого дня месяца)

Начало инкрементального периода:

21.01.2025 - (3 - 1) мес = 01.11.2024

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

для архивного периода:

select * from test_incremental where datekey >= date(\'2024-10-01\') and datekey < date(\'2024-11-01\')

для инкрементального периода:

select * from test_incremental where datekey >= date(\'2024-11-01\') and datekey < date(\'2024-12-01\')

select * from test_incremental where datekey >= date(\'2024-12-01\') and datekey < date(\'2025-01-01\')

select * from test_incremental where datekey >= date(\'2025-01-01\') and datekey < date(\'2025-02-01\')

Смотрите также

Загрузка данных

На этой странице

1 Об инкрементальной загрузке данных
2 Настройка инкрементальной загрузки данных
3 Внутренняя логика работы инкрементальной загрузки

Время чтения: 2 мин.

Нужна дополнительная помощь?

Свяжитесь с технической поддержкой.