문을 사용해 파이프라인의 CREATE FLOW 테이블에 대한 흐름을 생성하거나 백필을 만듭니다.
Note
CREATE FLOW 스트리밍 테이블을 타겟팅하는 것은 두 가지 모두 AUTO CDC ... INTOREPLACE WHERE 와 플로우를 지원합니다. 관리된 테이블은 CREATE TABLE ... FLOW 변경 데이터 캡처를 지원하지 않습니다: 관리 테이블에 대한 흐름은 AUTO CDC ... INTO 로 실패합니다 MANAGED_TABLE_DOES_NOT_SUPPORT_CDC.
CREATE FLOW 스트리밍 테이블로 들어가는 것은 그런 제한을 받지 않습니다. 참고 CREATE TABLE ... 흐름(파이프라인).
Syntax
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
생성할 흐름의 이름입니다.
주석
흐름에 대한 선택적 설명입니다.
-
AUTO CDC ... INTO을 사용하여 흐름을 정의하는 문입니다create_auto_cdc_flow_spec.AUTO CDC ... INTO문 또는INSERT INTO문 중 하나를 포함해야 합니다. 데이터 변경 의미 체계를 소스 쿼리에서 사용할 때AUTO CDC ... INTO를 사용합니다.자세한 내용은 AUTO CDC INTO(파이프라인)를 참조하세요.
자동 CDC ... 스냅샷에서
AUTO CDC ... INTO변경 피드를 읽는 대신 스냅샷을 비교하여 변경 사항을 도출하는 문장입니다. 소스 데이터에서 변경 데이터 캡처가 활성화되어 있지 않고 전체 스냅샷만 사용할 수 있을 때는 이 양식을 사용하세요. 소스는 두 부분으로 지정되어 있습니다: 스냅샷 데이터를 읽는 필수FROM SNAPSHOT (snapshot_query)절과 다음에 처리할 스냅샷 버전을 선택하는 선택적WITH VERSION (version_query)절입니다. 스냅샷의 자동 CDC 작동 방식을 참조하세요.스냅샷에서 (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행을 반환할 때는 값이 null이 아님을 의미합니다. 열은 스칼라 값일 수도 있는데, 예를 들어 ,
BIGINTSTRUCT모든 필드가 순서 가능할 수 있습니다. 두 개 이상의 열, 여러 행, 또는 null 값을 반환하는 버전 쿼리는 로 인해 흐름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)나열된 열을 제외한 모든 열을 추적하도록 하세요.
스냅샷 CDC는 또는
SEQUENCE BY.을 지원WHERE하지 않습니다. 교차 스냅샷 순서는 로 표현WITH VERSION (...)됩니다.target_table
업데이트할 테이블입니다. 스트리밍 테이블이어야 합니다.
INSERT 로
대상 테이블에 삽입되는 테이블 쿼리를 정의합니다.
ONCE옵션이 제공되지 않으면 쿼리는 스트리밍 쿼리여야 합니다. 소스에서 읽기 위해 스트리밍 의미론을 사용하려면 STREAM 키워드를 사용하세요. 기존 레코드에 대한 변경이나 삭제가 발생하면 읽기 작업 중 오류가 발생합니다. 가장 안전한 방법은 정적 또는 추가 전용 소스에서 읽는 것입니다. 변경 커밋이 포함된 데이터를 수집하려면, Python과skipChangeCommits옵션을 사용하여 오류를 처리할 수 있습니다.INSERT INTO는AUTO CDC ... INTO와 서로 배타적입니다. 원본 데이터에 CDC(변경 데이터 캡처) 기능이 포함된 경우에 사용합니다AUTO CDC ... INTO. 원본이 사용하지 않는 경우 사용합니다INSERT INTO.스트리밍 데이터에 대한 자세한 내용은 파이프라인 사용하여 데이터 변환참조하세요.
사용 대체 ( column_name [, ...] ) sequence_column별 순서
Important
이 기능은 베타 버전으로 제공됩니다. Databricks 런타임 18.2 이상이 필요합니다.
플로우를 플로우로
REPLACE USING정의하며, 지정된 키 열과 일치하는 대상 테이블의 모든 행을 교체하고 나머지 행은 그대로 둡니다. 소스가 열별로 키잉된 부분 스냅샷 연속일 때 사용REPLACE USING하세요.SEQUENCE BY업데이트의 순서를 정해, 키의 가장 높은 시퀀스가 승리하도록 명령합니다. 업데이트가 순서가 뒤틀려 도착하더라도 말이죠.최소 하나의 키 열과 정확히 하나의
SEQUENCE BY열을 지정하세요. 쿼리는 스트리밍 쿼리여야 하며,BY NAME필수입니다.REPLACE USING와 결합ONCEAUTO CDC ... INTO하거나 결합할 수 없습니다.자세한 내용은 REPLACE USING flows로 된 부분 스냅샷 대체를 참조하세요.
교체 WHERE 조건
플로우를 플로우로
REPLACE WHERE정의하며, 대상 테이블의 목표 부분집합을 재계산하고 덮어쓰는 방식입니다. 각 업데이트마다 대상 테이블 매칭condition내 모든 행이 삭제되고, 동일한 술어 범위에 대해 소스 쿼리가 재계산되며, 결과가 삽입됩니다. 일치condition하지 않는 행은 손대지 않고 남겨둡니다. 소스 쿼리에 술어를 추가할 필요는 없습니다; 파이프라인 엔진이 소스에서 읽을 때 자동으로 적용합니다.REPLACE WHERE배치 의미론을 사용하기 때문에 소스 쿼리가 스트리밍 쿼리일 필요는 없습니다.BY NAME은 필수입니다.REPLACE WHERE,REPLACE USING, ,AUTO CDC ... INTO와 결합ONCE할 수 없습니다.자세한 내용은 REPLACE WHERE 플로우를 이용한 배치 처리 및 독립형 스트리밍 테이블에 대한 REPLACE WHERE 플로우를 참조하세요.
한번
필요에 따라 흐름을 백필과 같은 일회성 흐름으로 정의합니다.
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의 진행 중인 데이터를 결합하는 방법에 대해 더 알고 싶다면, 파이프라인을 이용한 과거 데이터 대체 자료를 참조하세요.