Расширенные возможности Table Stream
Предполагается, что вы уже прочитали Основы Table Stream; в примерах на этой странице используются та же таблица orders и тот же orders_stream.
Сценарии использования
| Что вы хотите сделать | Раздел |
|---|---|
| Узнать, до какого момента обработана каждая партиция и сколько изменений ожидает обработки; обрабатывать только активные партиции большой таблицы | Смещения обработки на уровне партиций |
| Сверить данные перед обработкой или восстановить повреждённую целевую таблицу | Чтение снимков @snapshot() |
| Полностью перезагрузить данные в целевую таблицу, а затем продолжить инкрементальную обработку с текущего момента | Сброс @reset() |
| Соединить изменения заказов с таблицей пользователей, согласовав состояние таблицы измерений с периодичностью обработки | Соединение с таблицами измерений |
| Передавать данные нескольким системам, обрабатывать несколько таблиц одним оператором или выполнять обработку параллельно | Несколько объектов Stream и параллельная обработка |
| Узнать, что происходит со Stream при добавлении столбцов или партиций в базовую таблицу, а также при её очистке, замене или удалении | Влияние изменений базовой таблицы |
| Настроить оповещения о накоплении необработанных изменений или восстановить обработку после сбоя задания | Мониторинг и восстановление |
| Найти причину ошибки и способ её устранения по сообщению | Распространённые ошибки |
Смещения обработки на уровне партиций
Смещения Stream хранятся отдельно для каждой партиции базовой таблицы, и каждая строка information_schema.table_stream_consumption соответствует одной партиции:
SELECT UNIT, CONSUMPTION_STATUS, LAG, LAST_CONSUMPTION_TIME
FROM information_schema.table_stream_consumption
WHERE DB_NAME = 'demo' AND STREAM_NAME = 'orders_stream'
ORDER BY UNIT;
+-----------+--------------------+-----------+-----------------------+
| UNIT | CONSUMPTION_STATUS | LAG | LAST_CONSUMPTION_TIME |
+-----------+--------------------+-----------+-----------------------+
| p20260912 | 469015470080000000 | 0 | 1789180806000 |
| p20260913 | 469041123123200000 | 0 | 1789267206000 |
| p20260914 | 469067681628160003 | 262144000 | 1789351206000 |
| p20260915 | N/A | 0 | -1 |
+-----------+--------------------+-----------+-----------------------+
| Столбец | Описание |
|---|---|
UNIT | Единица обработки, то есть имя партиции. У таблицы без явно заданных партиций есть единственная партиция с именем таблицы |
CONSUMPTION_STATUS | Временная метка фиксации (TSO), до которой обработана партиция. N/A означает, что партиция ещё ни разу не обрабатывалась |
LAG | Разница между последней зафиксированной TSO партиции и TSO, до которой выполнена обработка. 0 означает отсутствие необработанных изменений; N/A означает, что в партиции есть данные, но она ещё ни разу не обрабатывалась |
LAST_CONSUMPTION_TIME | Время последней обработки партиции (временная метка в миллисекундах); -1 означает, что партиция ещё ни разу не обрабатывалась |
Старшие биты TSO содержат физическое время в миллисекундах, а младшие 18 бит — логический счётчик, поэтому LAG, делённый на 262144 (2 в степени 18), приблизительно соответствует отставанию в миллисекундах. В примере выше p20260914 отстаёт примерно на 1 секунду.
Обработка по партициям
Когда изменения большой таблицы сосредоточены в нескольких активных партициях, используйте предложение PARTITION, чтобы обрабатывать только эти партиции; при этом продвигаются только смещения обработанных партиций:
INSERT INTO dwd_orders
SELECT order_id, status, amount
FROM orders_stream PARTITION (p20260914)
WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('APPEND', 'UPDATE_AFTER');
Несколько обращений к разным партициям одного Stream в одном операторе объединяются в единое обновление смещений.
В режиме с разделением вычислений и хранения один INSERT не может обрабатывать больше партиций, чем допускает параметр конфигурации FE cloud_table_stream_max_partitions_per_insert (по умолчанию 10000). При превышении лимита возникает ошибка Cloud Table Stream consumes N partitions, exceeding cloud_table_stream_max_partitions_per_insert=...; выполняйте обработку порциями с помощью PARTITION.
Добавление и удаление партиций
- Все данные партиций, добавленных после создания Stream (в том числе созданных динамическим и автоматическим партиционированием), выдаются как изменения; при первой обработке такая партиция читается с начала.
- После удаления партиции FE периодически удаляет запись о её смещении (
table_stream_partition_offset_cleanup_interval_second, по умолчанию 3600 секунд); никаких действий вручную не требуется. Необработанные изменения удалённой партиции исчезают вместе с ней.
Чтение снимков @snapshot()
<stream>@snapshot() возвращает снимок базовой таблицы на момент, соответствующий текущему смещению обработки Stream, то есть состояние базовой таблицы до изменений, ожидающих обработки в этом цикле:
SELECT order_id, status, amount FROM orders_stream@snapshot() ORDER BY order_id;
Рассмотрим данные из раздела Основы Table Stream: при создании Stream базовая таблица содержит заказы 1, 2 и 3, затем заказ 1 обновляется, заказ 2 удаляется и вставляется заказ 4, при этом обработка ещё не выполнялась. @snapshot() по-прежнему возвращает заказы 1 (до обновления), 2 и 3, а обычное чтение возвращает эти изменения. Соотношение следующее: снимок + ожидающие обработки изменения = текущее состояние базовой таблицы.
Особенности @snapshot():
- Не продвигает смещение, даже если результат записывается в другую таблицу с помощью
INSERT INTO ... SELECT FROM <stream>@snapshot(). - Возвращает снимок таблицы без виртуальных столбцов
__DORIS_STREAM_*. - Снимок формируется отдельно для каждой партиции на момент, соответствующий её собственному смещению обработки.
- Для Stream с
show_initial_rows = trueснимок партиций, которые ещё ни разу не обрабатывались, пуст (обработка начинается с «ничего», а все существующие данные относятся к ожидающим обработки изменениям).
Типичные сценарии:
| Сценарий | Способ использования |
|---|---|
| Сверка данных | Перед обработкой сравните результат @snapshot() с целевой таблицей, чтобы убедиться, что она соответствует состоянию базовой таблицы после предыдущей обработки, затем обработайте изменения текущего цикла |
| Восстановление целевой таблицы | Если целевая таблица повреждена, восстановите её из @snapshot() до состояния, соответствующего смещению обработки, затем продолжайте обычную обработку; изменения после этого смещения не будут ни пропущены, ни обработаны повторно |
| Снимок состояния «до изменений» в инкрементальных соединениях | См. ниже раздел Соединение с таблицами измерений |
Сброс @reset()
<stream>@reset() возвращает полный снимок текущего состояния базовой таблицы. Чтение обычным SELECT лишь показывает данные; обработка с помощью INSERT INTO ... SELECT FROM <stream>@reset() продвигает смещения всех партиций Stream до текущего момента, после чего Stream выдаёт только изменения, произошедшие после этой обработки:
-- Clear downstream, reload it in full, and advance the offset to now
TRUNCATE TABLE dwd_orders;
INSERT INTO dwd_orders
SELECT order_id, status, amount FROM orders_stream@reset();
-- Continue with incremental consumption
SELECT COUNT(*) FROM orders_stream; -- 0
Типичные сценарии:
- Целевой таблице нужна повторная полная синхронизация с последующей инкрементальной обработкой с текущего момента.
- Stream создан с
show_initial_rows = false, но существующие данные всё же требуется синхронизировать. - В Stream накопилось слишком много необработанных изменений, а целевой системе больше не нужна история изменений; можно сразу перейти к текущему моменту.
Как и @snapshot(), @reset() не предоставляет виртуальных столбцов __DORIS_STREAM_*.
Сравнение трёх форм чтения
<stream> | <stream>@snapshot() | <stream>@reset() | |
|---|---|---|---|
| Что возвращает | Изменения после смещения обработки | Снимок базовой таблицы на момент, соответствующий смещению обработки | Полный снимок текущего состояния базовой таблицы |
Продвигает ли обычный SELECT смещение | Нет | Нет | Нет |
Продвигает ли INSERT INTO ... SELECT смещение | Да, до верхней границы чтения | Нет | Да, для всех партиций до текущего момента |
Виртуальные столбцы __DORIS_STREAM_* | Есть | Нет | Нет |
| Типичные сценарии | Инкрементальная обработка | Сверка данных, восстановление целевой таблицы, снимок состояния «до изменений» в инкрементальных соединениях | Полная перезагрузка данных, пропуск накопленных изменений |
Соединение с таблицами измерений
Stream можно напрямую соединять с другими таблицами; таблица измерений читается в её текущем состоянии:
INSERT INTO dwd_orders_wide
SELECT o.order_id, o.status, o.amount, u.user_name, u.city
FROM orders_stream AS o
JOIN users AS u ON o.user_id = u.user_id
WHERE o.__DORIS_STREAM_CHANGE_TYPE_COL__ IN ('APPEND', 'UPDATE_AFTER');
Если сама таблица измерений меняется и требуется согласовать её состояние с периодичностью обработки, создайте Stream и для таблицы измерений и используйте его в том же задании обработки:
| Форма чтения | Какое состояние таблицы измерений возвращается |
|---|---|
users_stream@snapshot() | Снимок таблицы измерений на момент предыдущей обработки |
users_stream | Изменения таблицы измерений с момента предыдущей обработки |
users_stream@reset() | Снимок текущего состояния таблицы измерений |
Например, соедините изменения заказов с информацией о пользователях на момент предыдущей обработки, а изменения таблицы пользователей обработайте отдельно:
INSERT INTO dwd_orders_wide
SELECT o.order_id, o.status, o.amount, u.user_name, u.city
FROM orders_stream AS o
JOIN users_stream@snapshot() AS u ON o.user_id = u.user_id
WHERE o.__DORIS_STREAM_CHANGE_TYPE_COL__ IN ('APPEND', 'UPDATE_AFTER');
Этот оператор продвигает только смещение orders_stream; смещение users_stream продвигается оператором, который обрабатывает users_stream. Периодичность обработки обоих объектов Stream определяется оркестрацией ваших заданий.
Несколько объектов Stream и параллельная обработка
| Сценарий | Поведение | Рекомендация |
|---|---|---|
| Несколько объектов Stream для одной базовой таблицы | Каждый Stream хранит собственные смещения, которые не влияют друг на друга | Каждая целевая система может создать свой Stream и обрабатывать его в собственном темпе |
| Один оператор обрабатывает несколько объектов Stream | Например, изменения двух таблиц объединяются через UNION в одну целевую таблицу; смещения всех участвующих объектов Stream продвигаются в одной транзакции | - |
| Несколько потребителей одного Stream | Если два сеанса одновременно обрабатывают одну и ту же партицию одного Stream, транзакция, которая фиксируется позже, завершается ошибкой target offset already consumed и откатывается | У каждой партиции Stream должен быть только один потребитель в каждый момент времени; для параллельной обработки разделите задание по партициям или создайте отдельный Stream для каждого потребителя |
| Обработка одновременно с записью | Во время обработки Stream запись в базовую таблицу выполняется как обычно. Оператор обработки читает изменения, зафиксированные к моменту его запуска; записи, зафиксированные позже, остаются до следующего цикла | - |
Влияние изменений базовой таблицы
| Операция с базовой таблицей | Влияние на Stream |
|---|---|
ADD COLUMN / DROP COLUMN | Столбцы Stream синхронизируются автоматически. В последующих записях об изменениях новые столбцы содержат фактические значения; удалённые столбцы больше не выдаются |
ADD PARTITION, динамическое партиционирование, автоматическое партиционирование | Новая партиция обрабатывается с начала; все её данные выдаются как изменения |
DROP PARTITION | Необработанные изменения партиции исчезают вместе с ней; FE периодически удаляет запись о её смещении |
TRUNCATE TABLE / TRUNCATE PARTITION | Очищенные партиции получают новые идентификаторы партиций, а записанные после этого данные выдаются как изменения с начала; необработанные изменения, существовавшие до очистки, исчезают. Другие партиции не затрагиваются |
REPLACE PARTITION (замена временной партицией) | Заменяющая партиция обрабатывается с начала; необработанные изменения заменённой партиции исчезают |
RENAME таблицы или партиции | Не влияет; Stream ссылается на базовую таблицу по идентификатору |
REPLACE WITH TABLE | Stream продолжает ссылаться на исходный объект таблицы (который теперь носит другое имя), что обычно не соответствует ожидаемому поведению. После замены базовой таблицы удалите Stream и создайте его заново |
DROP TABLE для базовой таблицы | Stream сохраняется, но читать его больше нельзя. Удалите его с помощью DROP STREAM ... FORCE |
DROP DATABASE | Объекты Stream в этой базе данных удаляются вместе с ней. Объекты Stream в других базах данных, чьи базовые таблицы находились в удалённой базе данных, сохраняются, и для их удаления также требуется FORCE |
Операции DDL, запрещённые для базовой таблицы с changelog (например, MODIFY COLUMN), перечислены в разделе Changelog.
Мониторинг и восстановление
Мониторинг
Внутренние объекты Stream, создаваемые IVM, также отображаются в этих системных таблицах. Вы можете наблюдать за их состоянием и накоплением необработанных изменений, но не обрабатывайте, не сбрасывайте и не удаляйте их вручную; управляйте ими через задания обновления материализованных представлений и следуйте рекомендациям по эксплуатации IVM.
| Что отслеживать | Источник | Примечания |
|---|---|---|
| Накопленные необработанные изменения | LAG в information_schema.table_stream_consumption | Регулярно запрашивайте значение и настраивайте оповещения для партиций, в которых оно остаётся ненулевым или продолжает расти |
| Состояние | ENABLED, IS_STALE и STALE_REASON в information_schema.table_streams | Показывают состояние Stream, но в текущей реализации не сообщают об истечении TTL. Сопоставляйте задержку обработки со значением changelog.ttl_seconds базовой таблицы и проверяйте наличие ошибок об истёкших смещениях; см. Хранение и очистка |
| История обработки | LAST_CONSUMPTION_TIME в information_schema.table_stream_consumption | Позволяет проверить, выполняется ли задание обработки по расписанию |
Восстановление
| Ситуация | Действие |
|---|---|
| Оператор обработки завершился ошибкой | Смещение не изменилось; выполните оператор повторно, пока необходимые изменения ещё находятся в пределах срока хранения |
Срок хранения необходимых изменений истёк согласно changelog.ttl_seconds | Полностью перезагрузите данные в целевую таблицу с помощью @reset() и согласуйте смещения. Чтение min_delta завершается ошибкой при истёкших смещениях; detail и append_only пропускают изменения с истёкшим сроком хранения |
| Обработка завершилась успешно, но планировщик не зафиксировал это | Повторный запуск читает следующую порцию изменений и не обрабатывает предыдущую повторно. Чтобы полностью исключить побочные эффекты повторных запусков, используйте для целевой таблицы модель Unique Key (записи по первичному ключу по своей природе идемпотентны) |
| Данные в целевой таблице неверны и требуют восстановления | Восстановите состояние, соответствующее смещению обработки, с помощью @snapshot() и продолжите инкрементальную обработку либо полностью перезагрузите данные с помощью @reset(), что также продвинет смещение до текущего момента |
| Перезапуск FE или смена ведущего узла | Смещения сохраняются в метаданных, и после перезапуска обработка продолжается |
Распространённые ошибки
| Сообщение об ошибке | Причина | Действие |
|---|---|---|
Table Stream is experimental. Please set enable_table_stream=true to enable it. | На FE не включён enable_table_stream | Измените fe.conf и перезапустите FE |
Insert plan with Table stream failed. should enable changelog feature in FE config. | На FE не включён enable_feature_changelog | Измените fe.conf и перезапустите FE |
Base Olap table ... need to enable changelog for table stream | Для базовой таблицы не включён Changelog | Создайте базовую таблицу заново с "changelog.enable" = "true" |
MIN_DELTA table stream requires base mow table to enable changelog.need_historical_value=true | min_delta (тип по умолчанию) требует снимков состояния до изменений | Создайте базовую таблицу заново с включённым changelog.need_historical_value или используйте append_only |
not supported type: xxx | Недопустимое значение type | Используйте append_only / min_delta / detail |
target offset already consumed | Параллельная обработка одной партиции; смещение было продвинуто до фиксации этой транзакции | Выполните оператор повторно, чтобы прочитать следующую порцию; избегайте одновременной обработки несколькими потребителями |
Cloud Table Stream consumption only supports a normal INSERT into a local OLAP table | В режиме с разделением вычислений и хранения обработка выполняется внутри явной транзакции или с Group Commit либо целевая таблица не является внутренней | Используйте обычный INSERT INTO ... SELECT |
Cloud Table Stream consumes N partitions, exceeding cloud_table_stream_max_partitions_per_insert=... | Слишком много партиций в одном операторе | Выполняйте обработку порциями с помощью PARTITION или увеличьте значение параметра конфигурации |
Cloud Table Stream only supports DROP STREAM ... FORCE | DROP STREAM без FORCE в режиме с разделением вычислений и хранения | Используйте DROP STREAM ... FORCE |
Unknown column '__DORIS_STREAM_CHANGE_TYPE_COL__' ... | Обращение к виртуальному столбцу при использовании @snapshot() / @reset() | При чтении снимков виртуальные столбцы недоступны |