CREATE TABLE ... FLOW(파이프라인)

Important

이 기능은 베타 버전으로 제공됩니다.

CREATE TABLE ... FLOW 문을 사용하여 하나 이상의 흐름으로 작성된 파이프라인에서 관리되는 테이블을 만듭니다.

Syntax

CREATE TABLE
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ flow_clause ]

table_specification
  ( { column_identifier column_type [column_properties] } [, ...]
    [ CONSTRAINT expectation_name EXPECT (expectation_expr)
        [ ON VIOLATION { FAIL UPDATE | DROP ROW } ] ] [, ...] )

table_clauses
  { PARTITIONED BY (col [, ...]) |
    CLUSTER BY clause |
    LOCATION path |
    COMMENT table_comment |
    TBLPROPERTIES clause |
    WITH { ROW FILTER clause } } [ ... ]

flow_clause
  FLOW [ IMPORT CHECKPOINT checkpoint_path ] INSERT [ONCE] BY NAME query

여러 원본을 하나의 관리되는 테이블로 팬하려면 CREATE FLOW(파이프라인)를 대상으로 하는 여러 흐름을 선언합니다.

CREATE FLOW flow_name AS INSERT INTO table_name BY NAME query

Parameters

  • table_name

    만들 관리되는 테이블의 이름입니다. 이름이 정규화되지 않은 경우 테이블은 파이프라인의 대상 스키마에 만들어집니다. 이름은 스트리밍 테이블에 아직 속하지 않아야 합니다.

  • 테이블_규격

    필요에 따라 열, 해당 형식, 속성 및 설명을 정의합니다. 생략하면 스키마가 흐름 쿼리에서 유추됩니다.

  • CONSTRAINT expectation_name 기대 (expectation_expr) [ 위반 시 { 실패 UPDATE | 행 삭제 } ]

    관리 테이블에 데이터 품질 기대치를 추가합니다. 이러한 데이터 품질 기대치는 시간에 따라 추적할 수 있으며, 파이프라인의 이벤트 로그를 통해 접근할 수 있습니다. FAIL UPDATE 예상으로 인해 테이블을 만들고 테이블을 새로 고치면 처리가 실패합니다. DROP ROW 기대치로 인해 기대치가 충족되지 않으면 전체 행이 삭제됩니다. 파이프라인 기대를 사용하여 데이터 품질을 관리하기를 참조하세요.

    expectation_expr 리터럴, 테이블 내 열 식별자, 그리고 결정론적이고 내장된 SQL 함수 또는 연산자로 구성될 수 있으며, 단:

    또한 expectation_expr에는 하위 쿼리가 포함되어서는 안 됩니다.

  • PARTITIONED BY(col [, ...])

    필요에 따라 열의 하위 집합으로 테이블을 분할합니다.

  • CLUSTER BY 절

    필요에 따라 테이블에서 액체 클러스터링을 사용하도록 설정합니다. 결합 PARTITIONED BY 할 수 없습니다.CLUSTER BY

  • 위치 경로

    테이블 데이터의 선택적 스토리지 위치입니다.

  • 코멘트 table_comment

    STRING 테이블을 설명하는 리터럴입니다.

  • TBLPROPERTIES 절

    필요에 따라 하나 이상의 사용자 정의 테이블 속성을 설정합니다.

  • WITH ROW FILTER 절

    테이블에 행 필터 함수를 추가합니다. 해당 테이블에 대한 이후 쿼리는 함수가 계산되는 행의 하위 집합을 TRUE받습니다.

  • FLOW INSERT [ONCE] BY NAME 쿼리

    결과 열을 테이블 열과 query별로 일치시켜 테이블에 결과를 삽입하는 추가 흐름을 정의합니다. query 는 일괄 처리 또는 스트리밍 원본을 참조할 수 있습니다. ONCE 는 모든 업데이트가 아닌 한 번 흐름(예: 백필)을 실행합니다. 명명된 각 흐름은 스트리밍 테이블에서와 동일한 FLOW INSERT BY NAME 파이프라인 업데이트당 입력을 정확히 한 번 처리합니다.

  • 수입 검문소 checkpoint_path

    STRING 기존 구조화 스트리밍 체크포인트로 가는 경로를 포함하는 리터럴로, 마이그레이션된 스트림이 소스를 재처리하지 않고 마지막으로 커밋된 오프셋에서 재개되도록 합니다. 체크포인트 가져오기는 베타 단계입니다. 구조 화된 스트리밍 체크포인트 마이그레이션을 참조하세요.

구조화된 스트리밍 체크포인트 마이그레이션

Important

체크포인트 가져오기는 베타 단계입니다.

기존 구조화 스트리밍 워크로드를 소스를 재처리하지 않고 파이프라인으로 마이그레이션하는 데 사용 IMPORT CHECKPOINT 하세요. 사용하는 구조화 스트리밍 쿼리(클라우드 저장소, Unity 카탈로그 볼륨, DBFS 경로 등)로 checkpointLocation 설정 checkpoint_path 하세요. 첫 번째 파이프라인 업데이트에서는 해당 체크포인트를 파이프라인의 관리 저장소로 클론합니다. 흐름은 마지막으로 커밋된 오프셋에서 다시 시작되며, 그 상태(예: 집계, 중복 제거 키, 워터마크)가 그대로 유지됩니다. 이후 파이프라인 업데이트는 플로우의 복제된 체크포인트를 사용합니다; 원래 체크포인트는 수정되지 않습니다.

Warning

파이프라인을 실행하기 전에 원래의 구조화된 스트리밍 쿼리를 멈추세요. 원래 구조화된 스트리밍 쿼리는 가져오기 후 재사용할 수 있지만, 체크포인트 상태를 관리하고 파이프라인과 쿼리가 동시에 같은 테이블에 쓰지 않도록 해야 합니다. 이는 중복 데이터가 발생할 수 있습니다.

구조화된 스트리밍 쿼리를 쿼리가 작성한 테이블에 계속 쓰는 파이프라인 플로우로 다시 만들고, 체크포인트를 가져오세요:

-- Use the existing table your Structured Streaming query writes to; CREATE TABLE adopts it.
CREATE TABLE streaming_table
-- Continue from the imported checkpoint instead of reprocessing the source.
FLOW IMPORT CHECKPOINT '/Volumes/my_catalog/my_schema/checkpoints/my_stream'
INSERT BY NAME
  -- The same source your original query read from.
  SELECT * FROM STREAM(source_table);

체크포인트는 첫 번째 파이프라인 업데이트에서만 불러옵니다. 이후 업데이트에서는 무시 IMPORT CHECKPOINT합니다. 완전 새로고침은 체크포인트를 다시 가져오지 않습니다. 새롭고 빈 체크포인트에서 시작해 소스를 재처리합니다.

Limitations

  • 관리되는 테이블은 CDC 변경 흐름을 지원하지 않습니다. AUTO CDC INTO관리되는 테이블에 대한 (SQL) 또는 apply_changes / apply_changes_from_snapshot (Python)가 실패합니다.MANAGED_TABLE_DOES_NOT_SUPPORT_CDC CDC 대상의 경우, CREATE FLOW(파이프라인)로 정의된 흐름을 가진 (파이프라인)을 사용CREATE STREAMING TABLE하세요.
  • 관리되는 테이블은 지원하지 FLOW ... REPLACE WHERE않습니다. FLOW INSERT BY NAME만 지원됩니다. 테이블의 목표 부분집합을 재계산하고 덮어쓰려면, 스트리밍 테이블을 만들고 CREATE FLOW(파이프라인)로 플로우를 REPLACE WHERE 정의하세요. REPLACE WHERE 흐름을 사용한 Batch 처리를 참조하세요.
  • 관리되는 테이블은 Unity 카탈로그를 사용하는 파이프라인에서만 지원됩니다. Hive 메타스토어는 지원되지 않습니다.
  • 관리되는 테이블에 대해 기존 스트리밍 테이블의 이름을 다시 사용할 수 없습니다. 먼저 스트리밍 테이블을 삭제하거나 문이 .와 함께 CANNOT_SWITCH_STREAMING_TABLE_TO_MANAGED_TABLE실패합니다.
  • IMPORT CHECKPOINT 는 오직 에서만 FLOW INSERT BY NAME지원됩니다.

예제

-- Create a managed table populated by an inline append flow from a streaming table
CREATE TABLE output
FLOW INSERT BY NAME SELECT * FROM STREAM(samples.tpch.orders);

-- Create a managed table that ingests files with schema inference and evolution
CREATE TABLE raw_data
FLOW INSERT BY NAME
  SELECT * FROM STREAM read_files('abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/base/path');

-- Create a partitioned managed table from a streaming source
CREATE TABLE events
PARTITIONED BY (bucket)
FLOW INSERT BY NAME
  SELECT id, bucket FROM STREAM read_files('abfss://my_path', format => 'json');

-- Create a managed table with liquid clustering
CREATE TABLE orders_clustered
CLUSTER BY (order_date, customer_id)
FLOW INSERT BY NAME
  SELECT
    o_orderkey   AS order_id,
    o_custkey    AS customer_id,
    o_orderdate  AS order_date,
    o_totalprice AS total_price
  FROM STREAM(samples.tpch.orders);

-- Create a managed table with a data quality expectation that drops violating rows
CREATE TABLE valid_events
  (CONSTRAINT positive_id EXPECT (id > 0) ON VIOLATION DROP ROW)
FLOW INSERT BY NAME
  SELECT id FROM STREAM read_files('s3://bucket/path', format => 'json');

추가 리소스

  • 에 마지막으로 업데이트됨