Потоки REPLACE WHERE для автономных потоковых таблиц

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

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

Requirements

Потоки REPLACE WHERE имеют следующие требования:

Databricks рекомендует каталог Unity и бессерверные вычисления. Добавочное обновление поддерживается только для бессерверных вычислений.

Когда следует использовать потоки REPLACE WHERE

Используйте потоки REPLACE WHERE для следующих сценариев:

Добавочная пакетная обработка без семантики потоковой передачи. Обработка новых строк в пакетах без управления понятиями потоковой передачи, такими как подложки.
Выборочная повторная обработка: пересчитывайте только те строки, которые соответствуют предикату, оставляя все остальные строки без изменений.
Сценарии, выходящие за рамки стандартных материализованных возможностей представления:
- Целевые таблицы с более длительным сроком хранения, чем у источника
- Предотвращение повторной компиляции при изменении таблицы измерений
- Эволюция схемы без повторной компиляции всей истории

Создание потока REPLACE WHERE

Используйте FLOW REPLACE WHEREусловие в одной строке с CREATE OR REFRESH STREAMING TABLE:

CREATE OR REFRESH STREAMING TABLE orders_enriched
SCHEDULE EVERY 1 DAY
FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
SELECT
  o.order_id,
  o.date,
  o.region,
  p.product_name,
  o.qty,
  o.price
FROM orders_fct o
JOIN product_dim p
  ON o.product_id = p.product_id;

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

В качестве альтернативы используйте длинный синтаксис CREATE FLOW для определения потоковой таблицы и процесса REPLACE WHERE в отдельных операторах:

CREATE STREAMING TABLE orders_enriched;

CREATE FLOW orders_enriched AS
INSERT INTO orders_enriched BY NAME
REPLACE WHERE date >= date_add(current_date(), -7)
SELECT
  o.order_id,
  o.date,
  o.region,
  p.product_name,
  o.qty,
  o.price
FROM orders_fct o
JOIN product_dim p
  ON o.product_id = p.product_id;

Обе формы эквивалентны. Для полного CREATE FLOW справочника см. CREATE FLOW (pipelines).

Вам не нужно добавлять предикат в исходный запрос. Механизм конвейера автоматически применяет это при чтении данных из источника.

Примечание.

BY NAME является обязательным. Он гарантирует соответствие столбцов по имени, а не по позиции.

Заполнение исторических данных

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

Переопределения предиката: повторно выполните исходный запрос потока для одноразового диапазона предиката с помощью REFRESH STREAMING TABLE ... WHERE. Используется, когда исторические данные приходят из того же источника, что и добавочные данные.
Инструкции DML: вставьте в целевую таблицу напрямую, обходя поток. Используется, если исторические данные находятся в другом источнике, чем добавочные данные.

Переопределения предикатов

Переопределите предикат потока REPLACE WHERE для одного обновления без изменения определения таблицы. Переопределения предиката являются однократными, применяются только к обновлению и не влияют на последующие запланированные обновления.

Используйте условие WHERE с REFRESH STREAMING TABLE:

REFRESH STREAMING TABLE orders_enriched WHERE date BETWEEN '2020-01-01' AND '2024-12-31';

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

переопределение можно объединить с SYNC или ASYNC. Например, чтобы запустить обновление в фоновом режиме и сразу получить ссылку на обновление конвейера:

REFRESH STREAMING TABLE orders_enriched WHERE date >= date_add(current_date(), -30) ASYNC;

Примечание.

Оператор WHERE поддерживается только для потоковой таблицы, созданной с помощью оператора FLOW REPLACE WHERE. Указание его в любой другой таблице потоковой передачи возвращает ошибку. См. REFRESH (MATERIALIZED VIEW или STREAMING TABLE).

Инструкции DML

Запустите инструкции DML непосредственно в целевой таблице для загрузки строк, поступающих из другого источника, отличного от потока:

INSERT INTO orders_enriched
SELECT *
FROM orders_enriched_legacy
WHERE date < '2025-01-01';

Поведение полного обновления

Полное обновление потока REPLACE WHERE повторно выполняет исходный запрос с использованием только определенного предиката потока. Он окончательно удаляет каждую строку в целевом объекте, которая не соответствует этой предикате, включая строки, переопределяющие предикаты или инструкции DML, ранее вставленные за пределы определенного диапазона предиката.

Предупреждение

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

REFRESH STREAMING TABLE orders_enriched FULL;

Чтобы предотвратить полное обновление таблицы, задайте для свойства pipelines.reset.allowedfalseтаблицы значение :

CREATE OR REFRESH STREAMING TABLE orders_enriched
  TBLPROPERTIES (pipelines.reset.allowed = 'false')
  FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
  ...

Добавочное обновление

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

Когда применяется инкрементальное обновление

Все из перечисленного ниже должно быть истинным:

Конвейер выполняется на бессерверных вычислениях.
Поддерживается структура запроса. Сведения о поддерживаемом наборе операторов см. в инкрементальном обновлении.
Предикат ссылается на базовые столбцы из исходной таблицы. Предикаты по производным значениям, таким как результаты агрегатных или оконных функций, нельзя перенести на источник, из-за чего становится недоступным инкрементное обновление.
Внешний DML не изменил строки в текущем окне замены. DML, изменяющий строки за пределами текущего окна, не затрагивается.
Текущее окно замены не включает строки, исключенные из предыдущего предиката. Если вы расширяете предикат так, чтобы он охватывал диапазон, который ранее не обрабатывался, это обновление выполняется с полным пересчётом. Последующие обновления имеют право на добавочное обновление снова.
Предикат детерминирован. Предикаты, использующие недетерминированные функции, такие как rand(), отключают инкрементное обновление. Темпоральные функции, такие как current_date() разрешены.

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

Диагностировать переход к полному пересчёту

Когда обновление переходит к полному пересчёту, причина указывается в событии planning_information для потока. См. объект IssueType в схеме журнала событий конвейера. В следующей таблице перечислены причины, сообщаемые в событии:

Причина	Meaning
`EXTERNAL_CHANGE_IN_REPLACE_WINDOW`	Внешняя операция DML изменила строки в текущем окне замены.
`REPLACE_WHERE_NOT_DETERMINISTIC`	Предикат использует недетерминированные выражения.
`PRIOR_REPLACE_WHERE_NOT_DETERMINISTIC`	Предыдущее обновление использовало недетерминированный предикат.
`PRIOR_REPLACE_WHERE_PREDICATE_MISSING`	Не удалось получить предикат, использованный в предыдущем обновлении. Это может произойти, если предикат был некорректно сериализирован во время того обновления.
`STALE_TARGET_DATA_IN_WINDOW`	Предикат менялся между обновлениями, и текущее окно включает строки, которые предыдущий предикат исключал.
`CANNOT_PUSH_PREDICATE`	Предикат нельзя передать вниз ни в один источник.
`REPLACE_WHERE_BACKFILL`	Обновление представляет собой бэкфилл, использующий одноразовое переопределение предиката для заполнения исторических данных или исправления предыдущих операций записи.

Examples

В следующих примерах показаны распространенные шаблоны потока REPLACE WHERE .

Пример 1. Сохранение статистических статистических данных из источника ограниченного хранения

Этот пример сохраняет ежедневные статистические данные неограниченное время даже после того, как необработанные данные выходят из исходной таблицы (3-дневное хранение):

CREATE OR REFRESH STREAMING TABLE events_agg
FLOW REPLACE WHERE date >= date_add(current_date(), -3) BY NAME
SELECT
  date,
  key,
  SUM(val) AS agg
FROM events_raw
GROUP BY ALL;

Пример 2. Предотвращение повторной компиляции при изменении таблицы измерений

В этом примере строки фактов сохраняются без изменений при изменении атрибутов измерения:

CREATE OR REFRESH STREAMING TABLE fact_dim_join
FLOW REPLACE WHERE date >= date_add(current_date(), -1) BY NAME
SELECT
  f.date,
  f.user_id,
  d.region,
  f.revenue
FROM fact_table f
JOIN dim_users d
  ON f.user_id = d.user_id;

Если регион пользователя изменяется, перекомпьютируются только последние строки. Исторические строки сохраняют значение региона во время их записи.

Пример 3. Добавление новой метрики без повторной компиляции полной истории

В этом примере показано, как изменить определение таблицы и заполучить только целевой диапазон:

Определите начальную таблицу:

CREATE OR REFRESH STREAMING TABLE clickstream_daily
FLOW REPLACE WHERE event_date >= date_add(current_date(), -7) BY NAME
SELECT
  event_date,
  page_id,
  COUNT(*) AS clicks
FROM clickstream_raw
GROUP BY ALL;

Обновите запрос, чтобы добавить uniq_users:

CREATE OR REFRESH STREAMING TABLE clickstream_daily
FLOW REPLACE WHERE event_date >= date_add(current_date(), -7) BY NAME
SELECT
  event_date,
  page_id,
  COUNT(*) AS clicks,
  COUNT(DISTINCT user_id) AS uniq_users
FROM clickstream_raw
GROUP BY ALL;

Строки старше 7-дневного окна содержат NULL для uniq_users.

Пример 4: Выполните итерации на небольшом временном диапазоне перед дозаполнением всей истории

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

Начните с короткого окна для проверки метрик и итерации бизнес-логики с более низкими затратами на вычисления:

CREATE OR REFRESH STREAMING TABLE revenue_attribution
FLOW REPLACE WHERE event_date >= date_add(current_date(), -7) BY NAME
SELECT
  event_date,
  campaign_id,
  SUM(revenue) AS total_revenue
FROM marketing_events
GROUP BY ALL;

Короткий интервал пересчитывает только последние 7 дней при каждом обновлении, поэтому корректируйте запрос столько раз, сколько необходимо, прежде чем запускать полный пересчёт за весь исторический период.

После завершения запроса используйте DML, чтобы заполнить полный исторический диапазон:

INSERT INTO revenue_attribution
SELECT
  event_date,
  campaign_id,
  SUM(revenue) AS total_revenue
FROM marketing_events
WHERE event_date < date_add(current_date(), -7)
GROUP BY ALL;

Обратная связь

Были ли сведения на этой странице полезными?

Последнее обновление: 2026-10-16