Инкрементальное чтение с Doris Changelog

Для таблицы Doris с включённым changelog ("changelog.enable" = "true") Flink Doris Connector поддерживает непрерывное чтение изменений на уровне строк. Для нового потребителя режим initial сначала считывает текущий снимок таблицы, а затем переключается на инкрементальное чтение. Изменения, необходимые для этого перехода, должны оставаться в пределах окна хранения changelog исходной таблицы.

Требования к версиям

Для этой функции требуются Flink Doris Connector 26.4.0 или новее и Doris 5.0.0 или новее. Doris changelog является экспериментальной функцией Doris 5.0.0, которую необходимо включить с помощью enable_feature_changelog = true в конфигурации FE. О том, как включить её для таблицы, какие модели таблиц поддерживаются и какие действуют ограничения, см. Changelog.

Пример

Сначала включите changelog при создании исходной таблицы Doris. Параметр changelog.need_historical_value необходим, если потребителю нужны значения обновляемых строк до изменения:

CREATE DATABASE IF NOT EXISTS test;

CREATE TABLE test.student_changelog_source (
    id INT,
    name VARCHAR(50),
    age INT
)
UNIQUE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES (
    "replication_num" = "1",
    "changelog.enable" = "true",
    "changelog.need_historical_value" = "true",
    "changelog.ttl_seconds" = "86400"
);

INSERT INTO test.student_changelog_source VALUES (1, 'Alice', 18);

Затем включите Flink Checkpoint и создайте таблицу источника Doris:

SET 'execution.checkpointing.interval' = '10s';

CREATE TABLE student_changelog (
    id INT,
    name STRING,
    age INT,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'doris',
    'fenodes' = '127.0.0.1:8030',
    'table.identifier' = 'test.student_changelog_source',
    'username' = 'root',
    'password' = '',
    'source.scan.mode' = 'initial'
);

SELECT * FROM student_changelog;

Режимы запуска

После запуска задания изменения в test.student_changelog_source непрерывно передаются в виде записей changelog Flink. Выберите режим запуска для нового потребителя с помощью source.scan.mode:

Режим	Поведение
snapshot	Считывает текущий снимок и останавливается. Это режим по умолчанию.
initial	Считывает текущий снимок и после завершения чтения снимка переключается на непрерывное чтение changelog Doris.
latest	Пропускает снимок и считывает изменения, созданные после запуска задания.
from-timestamp	Пропускает снимок и считывает изменения начиная с момента времени включительно, заданного в source.scan.timestamp в формате yyyy-MM-dd HH:mm:ss.

При возобновлении задания сохранённый прогресс чтения имеет приоритет над режимом запуска. Правила восстановления приведены в разделе Публикация прогресса чтения в Doris. Как для восстановленного смещения, так и для запуска в режиме from-timestamp действует срок хранения changelog.ttl_seconds исходной таблицы; поведение при истечении срока хранения описано ниже.

Типы изменений

По умолчанию Connector передаёт полные изменения строк в режиме detail. Установите для source.changelog.increment-type значение min_delta, чтобы получать минимальный набор изменений, или append_only, чтобы получать только события добавления.

Connector считывает изменения с помощью запросов Doris @incr. Если для запрошенной начальной позиции истёк срок хранения, заданный в changelog.ttl_seconds, поведение зависит от типа изменений:

Тип изменений	Поведение при истечении срока хранения для начальной позиции
detail (по умолчанию), append_only	Doris сдвигает фактическую начальную позицию к первому сохранённому TSO и пропускает изменения с истёкшим сроком хранения. Задание может продолжать работу без ошибки истечения срока хранения, но эти изменения будут отсутствовать в его выходных данных.
min_delta	Чтение завершается ошибкой Changelog offset has expired according to changelog.ttl_seconds, чтобы не возвращать неполные итоговые изменения.

Эти правила действуют при обычном чтении, после начального снимка и при восстановлении из сохранённого состояния. Работающее задание или успешный Checkpoint не гарантируют, что срок хранения всех изменений ещё не истёк. Параметр source.changelog.visible-wait-timeout не обеспечивает повторных попыток при ошибках истечения срока хранения.

Примечания

• Инкрементальное чтение использует Arrow Flight SQL. Connector включает его по умолчанию и автоматически определяет его порт.
• Включите Flink Checkpoint.
• Значение Doris changelog.ttl_seconds по умолчанию составляет 86400 секунд (одни сутки) и должно быть больше 0. Истечение срока хранения не зависит от задания Flink или его Checkpoint. Увеличение TTL не позволяет восстановить физически удалённые записи.
• Задайте TTL так, чтобы он покрывал длительность чтения начального снимка, накопившийся объём непрочитанных изменений и время, необходимое для восстановления после сбоя и чтения отставших данных, с дополнительным запасом. Отслеживайте возраст самых старых изменений, которые ещё нужны заданию, в том числе при восстановлении из более раннего Checkpoint или Savepoint.
• Для автоматической физической очистки также включите enable_feature_changelog = true на BE и оставьте автоматическую компакцию включённой. Инкрементальные запросы отфильтровывают записи с истёкшим сроком хранения ещё до завершения физической очистки. См. Хранение и очистка changelog.
• При инкрементальном чтении Doris ожидает завершения выполняющихся транзакций, затрагивающих исходную таблицу в пределах окна чтения. Если время ожидания истекает, чтение возвращает ошибку. Только при этой ошибке Connector повторяет чтение того же окна в течение времени, заданного в source.changelog.visible-wait-timeout (по умолчанию: 5m). Установите значение 0s, чтобы отключить повторные попытки Connector; при других ошибках чтение сразу завершается сбоем.

Восстановление после истечения срока хранения changelog

Если срок хранения необходимых изменений истёк, существующее смещение не позволяет выполнить полное инкрементальное восстановление. Если задание должно поддерживать согласованность данных в системах-получателях с текущей исходной таблицей, заново выполните начальную загрузку с source.scan.mode = 'initial', не восстанавливая Checkpoint или Savepoint с истёкшим сроком хранения, и перестройте данные в системах-получателях на основе нового снимка. Если настроена таблица смещений, также используйте новый source.changelog.consumer-id, для которого нет сохранённой записи, либо удалите оба параметра: source.changelog.offset-table и source.changelog.consumer-id. Иначе старое состояние Flink или внешнее смещение по-прежнему будет иметь приоритет над initial и не позволит получить новый полный снимок.

Новый снимок восстанавливает текущее состояние таблицы; он не может воспроизвести исторические события с истёкшим сроком хранения. Запуск в режиме latest или с более поздним временем в from-timestamp также пропускает отсутствующий интервал. Используйте эти режимы только в том случае, если пропуск этого интервала допустим или отсутствующие данные были восстановлены отдельно.

Семантика доставки и дедупликация

Источник Doris обеспечивает доставку с семантикой at-least-once при инкрементальном чтении, пока все необходимые записи изменений сохраняются в Doris. После сбоя события изменений могут передаваться повторно и влиять на результаты запросов Flink. Дедупликация событий CDC во Flink по умолчанию отключена. Чтобы использовать её, установите следующий параметр перед отправкой запроса:

SET 'table.exec.source.cdc-events-duplicate' = 'true';

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

Checkpoint сохраняют состояние чтения, а дедупликация CDC обрабатывает повторно переданные события. Эти механизмы не предотвращают истечение срока хранения changelog Doris и не восстанавливают события, уже утраченные из-за истечения срока хранения.

Публикация прогресса чтения в Doris (необязательно)

Необязательная таблица смещений сохраняет прогресс чтения в Doris для наблюдения и автоматического возобновления задания, которое не восстанавливается из Checkpoint или Savepoint.

Перед отправкой задания создайте следующую таблицу смещений. Учётной записи Doris, настроенной с помощью username и password, нужны права SELECT и INSERT для этой таблицы:

CREATE DATABASE IF NOT EXISTS ops;

CREATE TABLE ops.flink_source_offsets (
    consumer_id VARCHAR(256) NOT NULL,
    offset_timestamp DATETIME NOT NULL,
    update_time DATETIMEV2(3) NOT NULL
)
UNIQUE KEY(consumer_id)
DISTRIBUTED BY HASH(consumer_id) BUCKETS 1
PROPERTIES (
    "replication_num" = "1"
);

Оставьте Flink Checkpoint включённым и добавьте следующие параметры в предложение WITH таблицы источника из примера выше:

'jdbc-url' = 'jdbc:mysql://127.0.0.1:9030',
'source.changelog.offset-table' = 'ops.flink_source_offsets',
'source.changelog.consumer-id' = 'student-sync'

При перезапуске задания используйте ту же таблицу смещений и тот же source.changelog.consumer-id. Для разных заданий или исходных таблиц используйте разные идентификаторы, чтобы избежать перезаписи прогресса чтения.

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

После публикации прогресса запросите его в Doris:

SELECT consumer_id, offset_timestamp
FROM ops.flink_source_offsets
WHERE consumer_id = 'student-sync';

Пример результата:

+--------------+---------------------+
| consumer_id  | offset_timestamp    |
+--------------+---------------------+
| student-sync | 2026-10-09 10:00:00 |
+--------------+---------------------+

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

Условие	Поведение
Задание восстанавливается из Checkpoint или Savepoint	Используется сохранённое состояние чтения.
В таблице смещений есть запись для текущего consumer-id	Чтение возобновляется с offset_timestamp, переопределяя начальную позицию режима запуска. В режиме initial снимок также пропускается.
Таблица смещений не настроена или запись для потребителя отсутствует	Чтение начинается в соответствии с source.scan.mode.

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

Таблица смещений сохраняет только начальную позицию чтения, но не состояние последующих операторов Flink или приёмника. Чтобы восстановить это состояние, восстановите Checkpoint или Savepoint. О смещениях с истёкшим сроком хранения см. Восстановление после истечения срока хранения changelog.

Параметры

Следующие параметры управляют инкрементальным чтением. Остальные параметры источника перечислены в разделе Чтение данных из Doris.

Ключ	Значение по умолчанию	Обязательный	Описание
source.scan.mode	snapshot	N	Режим запуска источника. Поддерживаемые значения: snapshot, initial, latest и from-timestamp.
source.scan.timestamp	--	N	Обязателен для from-timestamp. Начальное время включительно в формате yyyy-MM-dd HH:mm:ss. Сохранённый прогресс чтения имеет приоритет; см. Правила восстановления.
source.changelog.increment-type	detail	N	Тип изменений: detail, min_delta или append_only. При истечении срока хранения для начальной позиции detail / append_only пропускают изменения с истёкшим сроком хранения, а min_delta завершается ошибкой.
source.changelog.poll-interval	10s	N	Интервал опроса новых изменений. Минимальное значение составляет 1 секунду.
source.changelog.visible-wait-timeout	5m	N	Максимальное время, в течение которого Connector повторяет чтение того же окна после того, как Doris вернёт ошибку истечения времени ожидания видимости транзакции. Установите значение 0s, чтобы отключить повторные попытки; отрицательные значения недопустимы.
source.changelog.offset-table	--	N	Таблица Doris для сохранения прогресса чтения и его возобновления. Создайте её заранее и укажите в формате database.table. Настраивайте вместе с source.changelog.consumer-id и jdbc-url; поддерживается только в инкрементальных режимах.
source.changelog.consumer-id	--	N	Уникальный идентификатор задания-потребителя в таблице смещений. Сохраняйте его неизменным при перезапуске того же задания.