CREATE STREAM

CREATE OR ALTER STREAM

Preview Feature — Open — 모든 계정에서 사용할 수 있어요. 이미 존재하지 않으면 새 스트림을 만들고, 존재하면 그 스트림을 문에 정의된 스트림으로 변환해요. CREATE OR ALTER STREAM 문은 CREATE STREAM 문의 구문 규칙을 따르며 ALTER STREAM 문과 동일한 제한 사항을 가집니다. 구문은 스트림이 생성되는 객체에 따라 달라집니다. 다음 예제는 테이블의 스트림을 보여줘요.

-- Table stream
CREATE OR ALTER STREAM <name>
  ON TABLE <table_name>
  [ APPEND_ONLY = TRUE | FALSE ]
  [ SHOW_INITIAL_ROWS = TRUE | FALSE ]
  [ COMMENT = '<string_literal>' ]

Required parameters (필수 파라미터)

• name — 스트림의 식별자(이름)를 지정하는 문자열이에요. 스트림이 생성되는 스키마 내에서 고유해야 합니다. 또한 식별자는 알파벳 문자로 시작해야 하고, 전체 식별자 문자열을 큰따옴표로 감싸지 않는 한 공백이나 특수 문자를 포함할 수 없어요 (예: "My object"). 큰따옴표로 감싼 식별자는 대소문자를 구분합니다. 자세한 내용은 Identifier requirements 문서를 참고하세요.
• table_name — 스트림이 변경을 추적하는 테이블(즉 소스 테이블)의 식별자(이름)를 지정하는 문자열이에요. 스트림을 쿼리하려면 기본 테이블에 대한 SELECT 권한이 필요합니다.
• external_table_name — 스트림이 변경을 추적하는 외부 테이블(즉 소스 외부 테이블)의 식별자(이름)를 지정하는 문자열이에요. 스트림을 쿼리하려면 기본 외부 테이블에 대한 SELECT 권한이 필요합니다.
• stage_name — 스트림이 디렉터리 테이블 변경을 추적하는 스테이지(즉 소스 디렉터리 테이블)의 식별자(이름)를 지정하는 문자열이에요. 스트림을 쿼리하려면 기본 스테이지에 대한 USAGE(외부 스테이지) 또는 READ(내부 스테이지) 권한이 필요합니다.
• view_name — 소스 뷰의 식별자(이름)를 지정하는 문자열이에요. 스트림은 뷰의 기본 테이블에 대한 DML 변경을 추적해요. 뷰의 스트림에 대한 자세한 내용은 Streams on views 문서를 참고하세요. 스트림을 쿼리하려면 뷰에 대한 SELECT 권한이 필요합니다.

Optional parameters (선택 파라미터)

• TAG ( ... ) — 태그 이름과 태그 문자열 값을 지정해요. 태그 값은 항상 문자열이며 최대 문자 수는 256이에요. 태그 지정 방법은 Tag quotas 문서를 참고하세요.
• COPY GRANTS — 다음 CREATE STREAM 변형 중 하나로 새 스트림을 만들 때 원래 스트림의 접근 권한을 유지하도록 지정해요. 파라미터는 OWNERSHIP을 제외한 모든 권한을 기존 스트림에서 새 스트림으로 복사합니다. 기본적으로 CREATE STREAM 명령을 실행하는 역할이 새 스트림을 소유해요.
• AT | BEFORE — 과거의 특정 시점/지점(Time Travel 사용)에 스트림을 만들어요. AT | BEFORE 절은 과거 데이터가 요청되는 시점을 결정합니다.

  참고 AT | BEFORE 절에 지정된 과거 시점에 소스 객체에 대한 변경 추적 데이터가 없으면 CREATE STREAM 문이 실패합니다. 변경 추적이 기록되기 전의 과거 시점에는 스트림을 만들 수 없어요.

• APPEND_ONLY = TRUE | FALSE — 추가 전용(append-only) 스트림인지 여부를 지정해요. 추가 전용 스트림은 행 삽입만 추적합니다. 업데이트와 삭제 작업(테이블 잘라내기 포함)은 기록되지 않아요. 예를 들어 테이블에 10행이 삽입되고 추가 전용 스트림의 오프셋이 진행되기 전에 그중 5행이 삭제되면, 스트림은 10행을 기록합니다. 이 유형의 스트림은 표준 스트림보다 쿼리 성능이 뛰어나며, 전적으로 행 삽입에 의존하는 ELT(추출·로드·변환) 및 유사한 시나리오에 매우 유용해요. 표준 스트림은 변경 집합의 삭제된 행과 삽입된 행을 조인해 어떤 행이 삭제되고 어떤 행이 업데이트되었는지 결정합니다. 추가 전용 스트림은 추가된 행만 반환하므로 표준 스트림보다 훨씬 성능이 좋을 수 있어요. 예를 들어 추가 전용 스트림의 행이 소비된 직후 소스 테이블을 잘라낼 수 있고, 기록된 삭제 작업은 다음에 스트림이 쿼리되거나 소비될 때 오버헤드에 기여하지 않습니다. 기본값: FALSE
• INSERT_ONLY = TRUE — 삽입 전용(insert-only) 스트림인지 여부를 지정해요. 삽입 전용 스트림은 행 삽입만 추적하며, 삽입된 집합에서 행을 제거하는 삭제 작업(즉 no-ops)은 기록하지 않습니다. 예를 들어 두 오프셋 사이에 외부 테이블이 참조하는 클라우드 스토리지 위치에서 File1이 제거되고 File2가 추가되면, 요청된 변경 간격 내에 File1이 추가되었는지와 무관하게 스트림은 File2의 행에 대한 기록만 반환해요. 표준 테이블의 CDC 데이터를 추적할 때와 달리 클라우드 스토리지 파일의 과거 기록에 대한 접근은 Snowflake에 의해 관리되거나 보장되지 않아요. 덮어쓰거나 추가된 파일은 기본적으로 새 파일로 처리됩니다. 기본값: FALSE
• SHOW_INITIAL_ROWS = TRUE | FALSE — 스트림이 처음 소비될 때 반환할 기록을 지정해요. TRUE면 스트림은 스트림이 생성된 순간 소스 객체에 존재했던 행만 반환합니다. 이 행들에서 METADATA$ISUPDATE 컬럼은 FALSE 값을 보여줘요. 이후에는 가장 최근 오프셋 이후의 소스 객체 DML 변경을 반환합니다(즉 일반적인 스트림 동작). 이 파라미터는 스트림의 소스 객체 내용으로 다운스트림 프로세스를 초기화하는 것을 가능하게 해요. 기본값: FALSE
• COMMENT = '<string_literal>' — 스트림에 대한 주석을 지정하는 문자열(리터럴)이에요. 기본값: 값 없음

Output (출력)

스트림의 출력은 소스 객체와 같은 컬럼에 다음 추가 컬럼들을 더한 것을 포함해요.

Access control requirements (접근 제어 요구 사항)

이 작업을 실행하는 데 사용되는 역할은 최소한 다음 권한을 보유해야 해요. 표준 테이블의 스트림/뷰의 스트림/디렉터리 테이블의 스트림/외부 테이블의 스트림 각각에 대한 권한이 필요합니다.

스키마의 객체를 조작하려면 상위 데이터베이스에 대한 권한과 상위 스키마에 대한 권한이 각각 최소 하나씩 필요합니다. 지정된 권한 집합으로 사용자 정의 역할을 만드는 방법은 Creating custom roles 문서를, 보안 객체에 대한 SQL 작업의 역할·권한 부여에 대한 일반적인 정보는 Overview of Access Control 문서를 참고하세요.

Usage notes (사용 참고 사항)

Examples (예제)

Creating a table stream

mytable 테이블에 스트림을 만들어 봅시다.

CREATE STREAM mystream ON TABLE mytable;

소스 테이블과 함께 Time Travel을 사용해 봅시다. 지정된 타임스탬프의 날짜·시간 이전에 존재했던 mytable 테이블에 스트림을 만들어 봅시다.

CREATE STREAM mystream ON TABLE mytable BEFORE (TIMESTAMP => TO_TIMESTAMP(40*365*86400));

지정된 타임스탬프의 날짜·시간에 정확히 존재했던 mytable 테이블에 스트림을 만들어 봅시다.

CREATE STREAM mystream ON TABLE mytable AT (TIMESTAMP => TO_TIMESTAMP_TZ('02/02/2019 01:02:03', 'mm/dd/yyyy hh24:mi:ss'));

5분 전에 존재했던 mytable 테이블에 스트림을 만들어 봅시다.

CREATE STREAM mystream ON TABLE mytable AT(OFFSET => -60*5);

같은 소스 테이블의 기존 스트림 oldstream과 같은 오프셋을 가진 mytable 테이블에 스트림을 만들어 봅시다.

CREATE STREAM mystream ON TABLE mytable AT(STREAM => 'oldstream');

기존 mystream 스트림을 다시 만들되 현재 오프셋을 유지해 봅시다.

CREATE OR REPLACE STREAM mystream ON TABLE mytable AT(STREAM => 'mystream');

지정된 트랜잭션의 변경을 제외하고 그 트랜잭션까지의 트랜잭션을 포함한 mytable 테이블에 스트림을 만들어 봅시다.

CREATE STREAM mystream ON TABLE mytable BEFORE(STATEMENT => '8e5d0ca9-005e-44e6-b858-a8f5b37c5726');

Creating a stream on a single-table view

myview 뷰에 스트림을 만들어 봅시다.

CREATE STREAM mystream ON VIEW myview;

추가 예제는 Stream examples 문서를 참고하세요.

Creating an insert-only stream on an external table

외부 테이블 스트림을 만들고, 외부 테이블 메타데이터에 추가된 기록을 추적하는 스트림의 CDC(변경 데이터 캡처) 기록을 쿼리해 봅시다.

-- Create an external table that points to the MY_EXT_STAGE stage.
-- The external table is partitioned by the date (in YYYY/MM/DD format) in the file path.
CREATE EXTERNAL TABLE my_ext_table (
  date_part date as to_date(substr(metadata$filename, 1, 10), 'YYYY/MM/DD'),
  ts timestamp AS (value:time::timestamp),
  user_id varchar AS (value:userId::varchar),
  color varchar AS (value:color::varchar)
) PARTITION BY (date_part)
  LOCATION=@my_ext_stage
  AUTO_REFRESH = false
  FILE_FORMAT=(TYPE=JSON);

-- Create a stream on the external table
CREATE STREAM my_ext_table_stream ON EXTERNAL TABLE my_ext_table INSERT_ONLY = TRUE;

-- Execute SHOW streams
-- The MODE column indicates that the new stream is an INSERT_ONLY stream
SHOW STREAMS;
+-------------------------------+------------------------+---------------+-------------+--------------+-----------+------------------------------------+-------+-------+-------------+
| created_on                    | name                   | database_name | schema_name | owner        | comment   | table_name                         | type  | stale | mode        |
|-------------------------------+------------------------+---------------+-------------+--------------+-----------+------------------------------------+-------+-------+-------------|
| 2020-08-02 05:13:20.174 -0800 | MY_EXT_TABLE_STREAM    | MYDB          | PUBLIC      | MYROLE       |           | MYDB.PUBLIC.EXTTABLE_S3_PART       | DELTA | false | INSERT_ONLY |
+-------------------------------+------------------------+---------------+-------------+--------------+-----------+------------------------------------+-------+-------+-------------+

-- Add a file named '2020/08/05/1408/log-08051409.json' to the stage using the appropriate tool for the cloud storage service.

-- Manually refresh the external table metadata.
ALTER EXTERNAL TABLE my_ext_table REFRESH;

-- Query the external table stream.
-- The stream indicates that the rows in the added JSON file were recorded in the external table metadata.
SELECT * FROM my_ext_table_stream;
+----------------------------------------+------------+-------------------------+---------+-------+-----------------+-------------------+-----------------+---------------------------------------------+
| VALUE                                  | DATE_PART  | TS                      | USER_ID | COLOR | METADATA$ACTION | METADATA$ISUPDATE | METADATA$ROW_ID | METADATA$FILENAME                           |
|----------------------------------------+------------+-------------------------+---------+-------+-----------------+-------------------+-----------------+---------------------------------------------|
| {                                      | 2020-08-05 | 2020-08-05 15:57:01.000 | user25  | green | INSERT          | False             |                 | test/logs/2020/08/05/1408/log-08051409.json |
|   "color": "green",                    |            |                         |         |       |                 |                   |                 |                                             |
|   "time": "2020-08-05 15:57:01-07:00", |            |                         |         |       |                 |                   |                 |                                             |
|   "userId": "user25"                   |            |                         |         |       |                 |                   |                 |                                             |
| }                                      |            |                         |         |       |                 |                   |                 |                                             |
| {                                      | 2020-08-05 | 2020-08-05 15:58:02.000 | user56  | brown | INSERT          | False             |                 | test/logs/2020/08/05/1408/log-08051409.json |
|   "color": "brown",                    |            |                         |         |       |                 |                   |                 |                                             |
|   "time": "2020-08-05 15:58:02-07:00", |            |                         |         |       |                 |                   |                 |                                             |
|   "userId": "user56"                   |            |                         |         |       |                 |                   |                 |                                             |
| }                                      |            |                         |         |       |                 |                   |                 |                                             |
+----------------------------------------+------------+-------------------------+---------+-------+-----------------+-------------------+-----------------+---------------------------------------------+

Creating a standard stream on a directory table

mystage라는 스테이지의 디렉터리 테이블에 스트림을 만들어 봅시다.

CREATE STREAM dirtable_mystage_s ON STAGE mystage;

스트림을 채우기 위해 디렉터리 테이블 메타데이터를 수동으로 새로 고쳐 봅시다.

ALTER STAGE mystage REFRESH;

스트림의 가장 최근 오프셋 이후에 파일이 하나 이상 추가된 후 스트림을 쿼리해 봅시다.

SELECT * FROM dirtable_mystage_s;

+-------------------+--------+-------------------------------+----------------------------------+----------------------------------+-------------------------------------------------------------------------------------------+-----------------+-------------------+-----------------+
| RELATIVE_PATH     | SIZE   | LAST_MODIFIED                 | MD5                              | ETAG                             | FILE_URL                                                                                  | METADATA$ACTION | METADATA$ISUPDATE | METADATA$ROW_ID |
|-------------------+--------+-------------------------------+----------------------------------+----------------------------------+-------------------------------------------------------------------------------------------+-----------------+-------------------+-----------------|
| file1.csv.gz      |   1048 | 2021-05-14 06:09:08.000 -0700 | c98f600c492c39bef249e2fcc7a4b6fe | c98f600c492c39bef249e2fcc7a4b6fe | https://myaccount.snowflakecomputing.com/api/files/MYDB/MYSCHEMA/MYSTAGE/file1%2ecsv%2egz | INSERT          | False             |                 |
| file2.csv.gz      |   3495 | 2021-05-14 06:09:09.000 -0700 | 7f1a4f98ef4c7c42a2974504d11b0e20 | 7f1a4f98ef4c7c42a2974504d11b0e20 | https://myaccount.snowflakecomputing.com/api/files/MYDB/MYSCHEMA/MYSTAGE/file2%2ecsv%2egz | INSERT          | False             |                 |
+-------------------+--------+-------------------------------+----------------------------------+----------------------------------+-------------------------------------------------------------------------------------------+-----------------+-------------------+-----------------+

CREATE OR ALTER STREAM

테이블에 새 스트림을 만들거나 기존 스트림의 주석을 업데이트해 봅시다.

CREATE OR ALTER STREAM mystream
  ON TABLE mytable
  COMMENT = 'Tracks DML changes on mytable';

더 알아보기 (Learn more)

• ALTER STREAM
• DROP STREAM
• SHOW STREAMS
• DESCRIBE STREAM