스트리밍 테이블 만들기

아티클
07/02/2024

적용 대상: 예(예)로 표시된 확인 Databricks SQL Databricks Runtime 13.3 LTS 이상

Important

이 기능은 공개 미리 보기 상태입니다.

스트리밍 또는 증분 데이터 처리를 추가로 지원하는 델타 테이블인 스트리밍 테이블을 만듭니다.

스트리밍 테이블은 Delta Live 테이블 및 Unity 카탈로그를 사용하는 Databricks SQL에서만 지원됩니다. 지원되는 Databricks 런타임 컴퓨팅에서 이 명령을 실행하면 구문만 구문 분석됩니다. SQL을 사용하여 Delta Live Tables 파이프라인 구현을 참조하세요.

구문

{ CREATE OR REFRESH STREAMING TABLE | CREATE STREAMING TABLE [ IF NOT EXISTS ] }
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ AS query ]

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

column_properties
  { NOT NULL |
    COMMENT column_comment |
    column_constraint |
    MASK clause } [ ... ]

table_clauses
  { PARTITIONED BY (col [, ...]) |
    COMMENT table_comment |
    TBLPROPERTIES clause |
    SCHEDULE [ REFRESH ] CRON cron_string [ AT TIME ZONE timezone_id ] |
    WITH { ROW FILTER clause } } [...]

매개 변수

리프레쉬

지정한 경우 쿼리에 정의된 원본에서 사용할 수 있는 최신 데이터로 테이블을 새로 고칩니다. 쿼리가 시작되기 전에 도착하는 새 데이터만 처리됩니다. 명령을 실행하는 동안 원본에 추가되는 새 데이터는 다음 새로 고침까지 무시됩니다.
IF NOT EXISTS

지정되고 동일한 이름의 테이블이 이미 존재하는 경우 명령문은 무시됩니다.

IF NOT EXISTS 와 함께 REFRESH사용할 수 없습니다. 즉 CREATE OR REFRESH TABLE IF NOT EXISTS , 허용되지 않습니다.
table_name

만들 테이블의 이름입니다. 이름에는 임시 사양이 포함되지 않아야 합니다. 이름이 정규화되지 않으면 테이블이 현재 스키마에 만들어집니다.
table_specification

이 선택적 절은 열 목록, 해당 형식, 속성, 설명 및 열 제약 조건을 정의합니다.

테이블 스키마에서 열을 정의하지 않으면 지정 AS query해야 합니다.
- column_identifier
  
  열의 고유한 이름입니다.
  - column_type
    
    열의 데이터 형식 을 지정합니다.
  - NOT NULL
    
    지정한 경우 열이 값을 허용하지 NULL 않습니다.
  - COMMENT column_comment
    
    열을 설명하는 문자열 리터럴입니다.
  - column_constraint
    
    Important
    
    이 기능은 공개 미리 보기 상태입니다.
    
    스트리밍 테이블의 열에 기본 키 또는 외래 키 제약 조건을 추가합니다. hive_metastore 카탈로그의 테이블에는 제약 조건이 지원되지 않습니다.
  - MASK 절
    
    Important
    
    이 기능은 공개 미리 보기 상태입니다.
    
    열 마스크 함수를 추가하여 중요한 데이터를 익명화합니다. 해당 열의 모든 향후 쿼리는 열의 원래 값 대신 열에 대해 해당 함수를 평가한 결과를 받게 됩니다. 이는 함수가 값을 수정할지 여부를 결정하기 위해 호출하는 사용자의 ID 및/또는 그룹 멤버 자격을 검사할 수 있는 세분화된 액세스 제어 목적에 유용할 수 있습니다.
  - 제약 조건 expectation_name 예상(expectation_expr) [ ON VIOLATION { FAIL UPDATE | DROP ROW } ]
    
    테이블에 데이터 품질 기대치를 추가합니다. 이러한 데이터 품질 기대치는 시간이 지남에 따라 추적되고 스트리밍 테이블의 이벤트 로그를 통해 액세스할 수 있습니다. FAIL UPDATE 테이블을 만들고 테이블을 새로 고치면 처리가 실패합니다. 기대치가 DROP ROW 충족되지 않으면 전체 행이 삭제됩니다.
    
    expectation_expr은 다음을 제외하고 리터럴, 테이블 내의 열 식별자 및 결정적 기본 제공 SQL 함수 또는 연산자로 구성될 수 있습니다.
    - 집계 함수
      - 분석 창 함수
      - 순위 창 함수
      - 테이블 반환 생성기 함수
    또한 expr에는 하위 쿼리가 포함되어서는 안 됩니다.
  - table_constraint
    
    Important
    
    이 기능은 공개 미리 보기 상태입니다.
    
    스트리밍 테이블에 정보 기본 키 또는 정보 외래 키 제약 조건을 추가합니다. hive_metastore 카탈로그의 테이블에는 키 제약 조건이 지원되지 않습니다.
table_clauses

필요에 따라 새 테이블에 대한 분할, 주석, 사용자 정의 속성 및 새로 고침 일정을 지정합니다. 각 하위 절은 한 번만 지정할 수 있습니다.
- PARTITIONED BY
  
  테이블을 분할할 테이블 열의 선택적 목록입니다.
- COMMENT table_comment
  
  STRING 테이블을 설명하는 리터럴입니다.
- TBLPROPERTIES
  
  선택적으로 하나 이상의 사용자 정의 속성을 설정합니다.
- 일정 [ 새로 고침 ] CRON cron_string [ 표준 시간대 timezone_id ]
  
  제공된 경우 스트리밍 테이블 또는 구체화된 뷰를 예약하여 지정된 석영 cron 일정으로 데이터를 새로 고칩니다. time_zone_values만 허용됩니다. AT TIME ZONE LOCAL은 지원되지 않습니다. 없는 경우 AT TIME ZONE 세션 표준 시간대가 사용됩니다. AT TIME ZONE 세션 표준 시간대가 설정되어 있지 않으면 오류가 throw됩니다. SCHEDULE 의미 체계는 .에 해당합니다 SCHEDULE REFRESH.
  
  Delta Live Tables 파이프라인 정의에는 구문을 사용할 SCHEDULE 수 없습니다.
  
  SCHEDULE 명령에서 CREATE OR REFRESH 절을 사용할 수 없습니다. 일정은 명령의 CREATE 일부로 제공할 수 있습니다. ALTER STREAMING TABLE을 사용하여 만든 후 스트리밍 테이블의 일정을 변경합니다.
- WITH ROW FILTER 절
  
  Important
  
  이 기능은 공개 미리 보기 상태입니다.
  
  테이블에 행 필터 함수를 추가합니다. 해당 테이블의 모든 이후 쿼리는 함수가 부울 TRUE로 평가되는 행의 하위 집합을 받게 됩니다. 이는 함수가 특정 행을 필터링할지 여부를 결정하기 위해 호출하는 사용자의 ID 및/또는 그룹 멤버 자격을 검사할 수 있는 세분화된 액세스 제어 목적에 유용할 수 있습니다.
AS query

이 절은 .의 query데이터를 사용하여 테이블을 채웁니다. 이 쿼리는 스트리밍 쿼리여야 합니다. 증분 방식으로 처리하려는 모든 관계로 STREAM 키워드를 추가하여 이 작업을 수행할 수 있습니다. a와 함께 table_specification 지정하면 지정된 테이블 스키마에 table_specification 반환된 모든 열이 포함되어야 합니다. 그렇지 않으면 오류가 발생합니다query.query 쿼리할 때 반환 값으로 table_specification 지정되었지만 반환 query null 되지 않은 모든 열입니다.

이 절은 Databricks SQL에서 만든 스트리밍 테이블에 필요하지만 델타 라이브 테이블에는 필요하지 않습니다. 이 절이 Delta Live Tables에 제공되지 않은 경우 DLT 파이프라인의 APPLY CHANGES 명령에서 이 테이블을 참조해야 합니다. Delta Live Tables에서 SQL을 사용하여 변경 데이터 캡처를 참조 하세요.

스트리밍 테이블과 다른 테이블 간의 차이점

스트리밍 테이블은 증가하는 데이터 세트를 처리할 때 각 행을 한 번만 처리하도록 설계된 상태 저장 테이블입니다. 대부분의 데이터 세트는 시간이 지남에 따라 지속적으로 증가하므로 스트리밍 테이블은 대부분의 수집 워크로드에 적합합니다. 스트리밍 테이블은 데이터 새로 고침 및 짧은 대기 시간이 필요한 파이프라인에 적합합니다. 스트리밍 테이블은 새 데이터가 도착할 때 결과를 증분 방식으로 계산할 수 있으므로 각 업데이트로 모든 원본 데이터를 완전히 다시 계산하지 않고도 결과를 최신 상태로 유지할 수 있으므로 대규모 변환에도 유용할 수 있습니다. 스트리밍 테이블은 추가 전용인 데이터 원본용으로 설계되었습니다.

스트리밍 테이블은 쿼리에 제공된 원본에서 사용할 수 있는 최신 데이터를 처리하는 추가 명령(예: REFRESH)을 허용합니다. 제공된 쿼리에 대한 변경 내용은 이전에 처리되지 않은 데이터를 호출 REFRESH하여 새 데이터에만 반영됩니다. 기존 데이터에도 변경 내용을 적용하려면 다음을 수행FULL REFRESH하려면 실행 REFRESH TABLE <table_name> FULL 해야 합니다. 전체 새로 고침은 원본에서 사용할 수 있는 모든 데이터를 최신 정의로 다시 처리합니다. 전체 새로 고침이 기존 데이터를 자르기 때문에 전체 데이터 기록을 유지하지 않거나 Kafka와 같은 짧은 보존 기간이 있는 원본에서는 전체 새로 고침을 호출하지 않는 것이 좋습니다. 원본에서 데이터를 더 이상 사용할 수 없는 경우 이전 데이터를 복구하지 못할 수 있습니다.

행 필터 및 열 마스크

Important

이 기능은 공개 미리 보기 상태입니다.

행 필터를 사용하면 테이블 검색이 행을 가져올 때마다 필터로 적용되는 함수를 지정할 수 있습니다. 이러한 필터를 통해 후속 쿼리는 필터 조건자가 true로 평가되는 행만 반환합니다.

열 마스크를 사용하면 테이블 검색이 행을 가져올 때마다 열 값을 마스킹할 수 있습니다. 해당 열과 관련된 모든 이후 쿼리는 열에 대한 함수를 평가하고 열의 원래 값을 대체한 결과를 받게 됩니다.

행 필터 및 열 마스크를 사용하는 방법에 대한 자세한 내용은 행 필터 및 열 마스크를 사용하여 중요한 테이블 데이터 필터링을 참조 하세요.

행 필터 및 열 마스크 관리

스트리밍 테이블의 행 필터 및 열 마스크는 문을 통해 CREATE OR REFRESH 추가, 업데이트 또는 삭제해야 합니다.

동작

정의자로 새로 고침: 또는 REFRESH 문이 스트리밍 테이블을 새로 고치면 CREATE OR REFRESH 행 필터 함수는 정의자의 권한(테이블 소유자)으로 실행됩니다. 즉, 테이블 새로 고침은 스트리밍 테이블을 만든 사용자의 보안 컨텍스트를 사용합니다.
쿼리: 대부분의 필터는 정의자의 권한으로 실행되지만 사용자 컨텍스트(예: 및IS_MEMBER)를 CURRENT_USER 확인하는 함수는 예외입니다. 이러한 함수는 호출자로 실행됩니다. 이 방법은 현재 사용자의 컨텍스트에 따라 사용자별 데이터 보안 및 액세스 제어를 적용합니다.

가시성

또는 INFORMATION_SCHEMA카탈로그 탐색기를 사용하여 DESCRIBE EXTENDED지정된 스트리밍 테이블에 적용되는 기존 행 필터 및 열 마스크를 검사합니다. 이 기능을 사용하면 사용자가 스트리밍 테이블에 대한 데이터 액세스 및 보호 조치를 감사하고 검토할 수 있습니다.

제한 사항

테이블 소유자만 스트리밍 테이블을 새로 고쳐 최신 데이터를 가져올 수 있습니다.
ALTER TABLE 명령은 스트리밍 테이블에서 허용되지 않습니다. 테이블의 정의 및 속성은 or ALTER STREAMING TABLE 문을 통해 CREATE OR REFRESH 변경해야 합니다.
시간 이동 쿼리는 지원되지 않습니다.
DML 명령을 통해 테이블 스키마를 발전시키는 INSERT INTOMERGE 것은 지원되지 않습니다.
스트리밍 테이블에서는 다음 명령이 지원되지 않습니다.
- CREATE TABLE ... CLONE <streaming_table>
- COPY INTO
- ANALYZE TABLE
- RESTORE
- TRUNCATE
- GENERATE MANIFEST
- [CREATE OR] REPLACE TABLE
델타 공유는 지원되지 않습니다.
테이블 이름을 바꾸거나 소유자를 변경하는 것은 지원되지 않습니다.
테이블 제약 조건(예: PRIMARY KEY FOREIGN KEY 지원되지 않음)
생성된 열, ID 열 및 기본 열은 지원되지 않습니다.

예제

-- Creates a streaming table that processes files stored in the given external location with
-- schema inference and evolution.
> CREATE OR REFRESH STREAMING TABLE raw_data
  AS SELECT * FROM STREAM read_files('abfss://container@storageAccount.dfs.core.windows.net/base/path');

-- Creates a streaming table that processes files with a known schema.
> CREATE OR REFRESH STREAMING TABLE csv_data (
    id int,
    ts timestamp,
    event string
  )
  AS SELECT *
  FROM STREAM read_files(
      's3://bucket/path',
      format => 'csv',
      schema => 'id int, ts timestamp, event string');

-- Creates a streaming table with schema evolution and data quality expectations.
-- The table creation or refresh fails if the data doesn't satisfy the expectation.
> CREATE OR REFRESH STREAMING TABLE avro_data (
    CONSTRAINT date_parsing (to_date(dt) >= '2000-01-01') ON VIOLATION FAIL UPDATE
  )
  AS SELECT *
  FROM STREAM read_files('gs://my-bucket/avroData');

-- Stores the data from Kafka in an append-only streaming table.
> CREATE OR REFRESH STREAMING TABLE firehose_raw
  COMMENT 'Stores the raw data from Kafka'
  TBLPROPERTIES ('delta.appendOnly' = 'true')
  AS SELECT
    value raw_data,
    offset,
    timestamp,
    timestampType
  FROM STREAM read_kafka(bootstrapServers => 'ips', subscribe => 'topic_name');

-- Read data from another streaming table scheduled to run every hour.
> CREATE STREAMING TABLE firehose_bronze
  SCHEDULE CRON '0 0 * * * ? *'
  AS SELECT
    from_json(raw_data, 'schema_string') data,
    * EXCEPT (raw_data)
  FROM STREAM firehose_raw;

-- Creates a streaming table with a column constraint
> CREATE OR REFRESH STREAMING TABLE csv_data (
    id int PRIMARY KEY,
    ts timestamp,
    event string
  )
  AS SELECT *
  FROM STREAM read_files(
      's3://bucket/path',
      format => 'csv',
      schema => 'id int, ts timestamp, event string');

-- Creates a streaming table with a table constraint
> CREATE OR REFRESH STREAMING TABLE csv_data (
    id int,
    ts timestamp,
    event string,
    CONSTRAINT pk_id PRIMARY KEY (id)
  )
  AS SELECT *
  FROM STREAM read_files(
      's3://bucket/path',
      format => 'csv',
      schema => 'id int, ts timestamp, event string');

-- Creates a streaming table with a row filter and a column mask
> CREATE OR REFRESH STREAMING TABLE masked_csv_data (
    id int,
    name string,
    region string,
    ssn string MASK catalog.schema.ssn_mask_fn
  )
  WITH ROW FILTER catalog.schema.us_filter_fn ON (region)
  AS SELECT *
  FROM STREAM read_files('s3://bucket/path/sensitive_data')

다음을 통해 공유

스트리밍 테이블 만들기

구문

매개 변수

스트리밍 테이블과 다른 테이블 간의 차이점

행 필터 및 열 마스크

행 필터 및 열 마스크 관리

동작

가시성

제한 사항

예제

피드백

피드백

추가 리소스

다음을 통해 공유

스트리밍 테이블 만들기

구문

매개 변수

스트리밍 테이블과 다른 테이블 간의 차이점

행 필터 및 열 마스크

행 필터 및 열 마스크 관리

동작

가시성

제한 사항

예제

관련된 문서

피드백

피드백

추가 리소스