СОЗДАЙТЕ ПОТОК (конвейеры)

Используйте оператор CREATE FLOW для создания потоков или дозаполнения для таблиц в конвейере.

Note

CREATE FLOW Таргетирование таблицы потоков поддерживает обе AUTO CDC ... INTO потоки и REPLACE WHERE потоков. Управляемая таблица, созданная с , CREATE TABLE ... FLOW не поддерживает сбор данных изменений: AUTO CDC ... INTO поток против управляемой таблицы не работает с MANAGED_TABLE_DOES_NOT_SUPPORT_CDC. CREATE FLOW Доступ к столу для стриминга не подпадает под такое ограничение. См.CREATE TABLE ... FLOW (трубопроводы).

Синтаксис

CREATE FLOW flow_name [COMMENT comment] AS
{
  AUTO CDC [ONCE] INTO target_table create_auto_cdc_flow_spec |
  AUTO CDC [ONCE] INTO target_table create_auto_cdc_from_snapshot_spec |
  INSERT [ONCE] INTO target_table BY NAME [ replace_using_spec | REPLACE WHERE condition ] query
}

create_auto_cdc_from_snapshot_spec
  FROM SNAPSHOT ( snapshot_query )
  [ WITH VERSION ( version_query ) ]
  KEYS ( key [, ...] )
  [ STORED AS { SCD TYPE 1 | SCD TYPE 2 } ]
  [ TRACK HISTORY ON { col_list | * EXCEPT ( col_list ) } ]

replace_using_spec
  REPLACE USING ( column_name [, ...] ) SEQUENCE BY sequence_column

Параметры

  • flow_name

    Имя создаваемого потока.

  • КОММЕНТАРИЙ

    Необязательное описание потока.

  • АВТОМАТИЧЕСКИЙ CDC В

    Оператор AUTO CDC ... INTO , определяющий поток с помощью оператора create_auto_cdc_flow_spec. Необходимо включить инструкцию AUTO CDC ... INTO или инструкцию INSERT INTO . Используется AUTO CDC ... INTO , когда исходный запрос использует семантику измененных данных.

    Дополнительные сведения см. в разделе AUTO CDC INTO (конвейеры).

  • AUTO CDC ... ИЗ СНИМКА

    Утверждение AUTO CDC ... INTO , которое выводит изменения путём сравнения снимков вместо чтения ленты изменений. Используйте эту форму, когда запись данных изменений не включена на исходном коде и доступны только полные снимки. Исходный код определяется в двух частях: обязательная FROM SNAPSHOT (snapshot_query) клауза, считывающая данные снимка, и необязательная WITH VERSION (version_query) клауза, выбирающая следующую версию снимка для обработки. Узнайте , как работает AUTO CDC FROM SNAPSHOT.

    • ИЗ СНИМКА (snapshot_query)

      Required. Запрос, который считывает данные снимка для выбранной WITH VERSION (...)версии . Движок дифференцирует результат с ранее зафиксированным снимком, чтобы получить вставки, обновления и удаления, а также объединяет их с целью, используя KEYS идентичность строки и STORED AS определяя, как сохраняются изменения.

      Вызовите current_snapshot_version() внутри этого запроса, чтобы ссылаться на выбранную версию WITH VERSION (...). Если WITH VERSION (...) не указано, current_snapshot_version() не вызывается внутри FROM SNAPSHOT (...).

      При WITH VERSION (...) пропуске движок считывает исходный код напрямую через FROM SNAPSHOT (...), и запрос к снимку выполняется только при начальной загрузке, в то время как целевой объект не имеет фиксированных данных и состояния снимка. При любом последующем обновлении, когда цель уже содержит данные или имеет состояние снимка, поток не заканчивается с AUTO_CDC_FROM_SNAPSHOT_NON_EMPTY_TARGET_WITHOUT_VERSION. Для обработки снимков после нескольких обновлений используйте WITH VERSION (...).

    • С ВЕРСИЕЙ (version_query)

      Optional. Запрос, который выбирает следующую версию снимка для обработки. Он должен вернуть ровно один столбец упорядочённого типа и либо 0, либо 1 строку. Когда она возвращает 1 строку, значение должно быть ненулевым. Столбец может быть скалярным значением, например BIGINT, или , STRUCT поля которого все упорядочиваемы. Запрос версии, возвращающий более одного столбца, более одной строки или нулевого значения, не проходит поток с INVALID_AUTO_CDC_FROM_SNAPSHOT_VERSION_QUERY.

      Во время одного обновления конвейера движок повторяет следующие шаги: он оценивает запрос версии; если запрос возвращает 0 строк, он останавливает обработку этого потока для текущего обновления; если запрос возвращает 1 строку, движок показывает это значение через current_snapshot_version(), оценивает запрос к снимку, коммитирует полученный снимок и открывает фиксированную версию через last_snapshot_version(). Затем движок переоценивает запрос версии, чтобы выбрать следующую. Одно обновление конвейера обрабатывает версии по порядку, пока запрос не возвращает строк.

      Каждая версия, возвращаемая после успешного коммита, должна быть больше ранее зафиксированной версии; не увеличивающая версия не проходит обновление с APPLY_CHANGES_FROM_SNAPSHOT_ERROR.OUT_OF_ORDER_SNAPSHOT_VERSION. Тип данных значения версии должен оставаться неизменным во всех коммитах снимков; изменение типа данных не завершает обновление с AUTO_CDC_FROM_SNAPSHOT_VERSION_SCHEMA_CHANGED. Полное обновление очищает состояние сохраненной версии.

    • КЛЮЧИ

      Required. Основные ключевые столбцы используются для идентификации строк между снимками с целью обнаружения изменений.

    • STORED AS { SCD TYPE 1 | SCD TYPE 2 }

      Optional. Указывает, как изменения хранятся в целевой таблице. Значение по умолчанию — SCD TYPE 1.

    • TRACK HISTORY ON { col_list | * EXCEPT (col_list) }

      Optional. Применимо только к SCD TYPE 2. Указывает, какие столбцы запускают новую строку истории при их изменении. Предоставьте либо явный список столбцов, либо * EXCEPT (col_list) отслеживайте все столбцы, кроме указанных.

    Snapshot CDC не поддерживает WHERE или SEQUENCE BY. Упорядочение перекрёстных снимков выражается через WITH VERSION (...).

  • target_table

    Таблица, которую нужно обновить. Это должна быть таблица потоковой передачи.

  • INSERT В

    Определяет запрос таблицы, вставленный в целевую таблицу. ONCE Если параметр не указан, запрос должен быть потоковым запросом. Используйте ключевое слово STREAM для использования семантики потоковой передачи для чтения из источника. Если чтение сталкивается с изменением или удалением существующей записи, возникает ошибка. Самое безопасное — читать из статических или источников только для добавления. Для приема данных с коммитами изменений можно использовать Python и skipChangeCommits опцию для обработки ошибок.

    INSERT INTO взаимоисключающ с AUTO CDC ... INTO. Используйте AUTO CDC ... INTO , когда исходные данные включают функции отслеживания измененных данных (CDC). Используйте INSERT INTO, когда источник этого не делает.

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

  • ЗАМЕНА С ( column_name [, ...] ) ПОСЛЕДОВАТЕЛЬНОСТЬ ПО sequence_column

    Important

    Эта функция доступна в бета-версии. Требуется Databricks Runtime 18.2 и выше.

    Определяет поток как REPLACE USING поток, который заменяет все строки в целевой таблице, соответствующие указанным ключевым столбцам, и оставляет все остальные строки нетронутыми. Используйте REPLACE USING тогда, когда исходный код — серия частичных снимков, ключевых по столбцам. SEQUENCE BY Упорядочивает обновления так, чтобы побеждает самая высокая последовательность для ключа, даже если обновления приходят не по порядку.

    Укажите хотя бы один ключевой столбец и ровно один SEQUENCE BY столбец. Запрос должен быть потоковым и BY NAME обязателен. REPLACE USINGнельзя комбинировать ни ONCEс AUTO CDC ... INTO .

    Для получения дополнительной информации см. раздел Частичная замена снимка с ЗАМЕНОЙ ИСПОЛЬЗОВАНИЕМ потоков.

  • Условие REPLACE WHERE

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

    REPLACE WHERE использует пакетную семантику, поэтому исходный запрос не обязательно должен быть стриминговым запросом. BY NAME является обязательным. REPLACE WHERE не комбинируется с ONCEпомощью , REPLACE USING, или AUTO CDC ... INTO.

    Для получения дополнительной информации см. разделы Пакетная обработка с REPLACE WHERE flows и REPLACE WHERE flows для автономных таблиц потоков.

  • ОДИН РАЗ

    При необходимости определите поток как единовременный поток, например восполнение. Использование ONCE меняет поток двумя способами:

    • Источник query или create_auto_cdc_flow_spec не является потоковой таблицей.
    • Поток выполняется один раз по умолчанию. Если конвейер обновляется с полным обновлением, ONCE поток снова запускается для повторного создания данных.

    ONCE Нельзя использовать с REPLACE USING, что требует источника потоковой передачи.

Примеры

-- EXAMPLE 1:
-- Create a streaming table, and add two flows that append data to it:
CREATE OR REFRESH STREAMING TABLE users;

-- first flow into target_table:
CREATE FLOW users_flow AS
INSERT INTO users BY NAME
SELECT * FROM stream(raw_data.users);

-- second flow into target_table:
CREATE FLOW backfill_users AS
INSERT ONCE INTO users BY NAME
SELECT * FROM user_backfill_table;

-- EXAMPLE 2:
-- Create a streaming table, and add a flow that applies CDC changes to it:
CREATE OR REFRESH STREAMING TABLE admins_cdc_target_table;

-- first flow into target_table:
CREATE FLOW admin_cdc_flow AS
AUTO CDC INTO admins_cdc_target_table
FROM stream(cdc_data.admins)
KEYS (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY sequenceNum
COLUMNS * EXCEPT (operation, sequenceNum)
STORED AS SCD TYPE 2;

-- EXAMPLE 3:
-- Create a streaming table, and add a REPLACE USING flow that keeps the latest
-- row for each payment_id from a stream of partial snapshots:
CREATE OR REFRESH STREAMING TABLE payments_latest;

CREATE FLOW payments_replace_flow AS
INSERT INTO payments_latest BY NAME
REPLACE USING (payment_id) SEQUENCE BY payment_date
SELECT payment_id, booking_id, status, payment_date
FROM STREAM(samples.wanderbricks.payments);

-- EXAMPLE 4:
-- AUTO CDC FROM SNAPSHOT without WITH VERSION: a one-time initial load from a snapshot table.
-- To process later snapshots on each update, add WITH VERSION (see EXAMPLE 5).
CREATE STREAMING TABLE users (user_id INT, name STRING, email STRING);

CREATE FLOW users_snapshot_flow AS
AUTO CDC ONCE INTO users
FROM SNAPSHOT (SELECT * FROM catalog.schema.users_snapshot)
KEYS (user_id)
STORED AS SCD TYPE 1;

-- EXAMPLE 5:
-- AUTO CDC FROM SNAPSHOT with WITH VERSION: pick the next file, then read it as the snapshot:
CREATE STREAMING TABLE orders (order_id INT, product STRING, quantity INT, order_date DATE);

CREATE FLOW orders_cdc AS
AUTO CDC INTO orders
FROM SNAPSHOT (
  SELECT order_id, product, quantity, order_date
  FROM read_files('/Volumes/catalog/schema/landing/orders/', format => 'json')
  WHERE _metadata.file_path = (SELECT version.path FROM current_snapshot_version())
)
WITH VERSION (
  SELECT struct(modification_time, path) AS version
  FROM list_files('/Volumes/catalog/schema/landing/orders/')
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR struct(modification_time, path) > (SELECT version FROM last_snapshot_version())
  )
  ORDER BY modification_time, path
  LIMIT 1
)
KEYS (order_id)
STORED AS SCD TYPE 2;

-- EXAMPLE 6:
-- One-time snapshot backfill plus a streaming CDC flow into the same target.
-- The backfill omits WITH VERSION, so it uses an implicit timestamp version. The
-- streaming flow's SEQUENCE BY column (event_ts) must be a TIMESTAMP so its type
-- matches that implicit version on the shared target.
CREATE STREAMING TABLE customers (
  customer_id INT, name STRING, email STRING, address STRING, event_ts TIMESTAMP
);

CREATE FLOW customers_snapshot_backfill AS
AUTO CDC ONCE INTO customers
FROM SNAPSHOT (SELECT * FROM catalog.schema.customers_snapshot)
KEYS (customer_id)
STORED AS SCD TYPE 1;

CREATE FLOW customers_cdc AS
AUTO CDC INTO customers
FROM STREAM(customers_cdc_events)
KEYS (customer_id)
SEQUENCE BY event_ts
STORED AS SCD TYPE 1;

-- EXAMPLE 7:
-- Create a streaming table, then add a REPLACE WHERE flow that recomputes and
-- overwrites a targeted window of the target table on each update:
CREATE STREAMING TABLE payments_latest;

CREATE FLOW payments_latest AS
INSERT INTO payments_latest BY NAME
REPLACE WHERE payment_date >= date_add(current_date(), -7)
SELECT payment_id, booking_id, status, payment_date
FROM samples.wanderbricks.payments;

Подробнее о сочетании одноразовой засыпки с текущим CDC на той же цели см. раздел «Заполнение исторических данных с конвейерами».

  • Последнее обновление: