튜토리얼: REST API로 Named Channel 시작하기

참고

가능하면 자동 배치와 더 간단한 통합의 이점을 얻기 위해 REST API 대신 Snowpipe Streaming SDK를 사용하세요. 환경에 SDK가 적합하지 않을 때 직접 REST를 사용하세요.

이 가이드는 Snowpipe Streaming REST API 와 SnowSQL로 생성된 JSON Web Token (JWT)을 사용해 Snowflake로 데이터를 스트리밍하는 방법을 보여줍니다. 순서가 보장되고 정확히 한 번 수집이 필요한 Named Channel REST 수집을 다룹니다.

더 간단한(대부분의 새 애플리케이션에 권장) Elastic Channel REST 경로는 튜토리얼: Elastic Channels 시작하기 (REST) 를 참고하세요.

출처: Snowflake 문서

본문

사전 요구 사항

시작하기 전에 다음 항목이 있는지 확인하세요.

Snowflake 사용자 및 객체:

키-쌍 인증으로 구성된 Snowflake 사용자. 다음 SQL 명령을 사용해 공용 키를 등록하세요.

ALTER USER MY_USER SET RSA_PUBLIC_KEY='<your-public-key>';

스트리밍 수집을 위한 Snowflake 데이터베이스, 스키마, 대상 테이블. 다음 SQL 명령을 사용하고 MY_DATABASE, MY_SCHEMA, MY_TABLE 같은 자리 표시자를 원하는 이름으로 바꿔 만들 수 있습니다.

-- Create Database and Schema
CREATE OR REPLACE DATABASE MY_DATABASE;
CREATE OR REPLACE SCHEMA MY_SCHEMA;

-- Create Target Table
CREATE OR REPLACE TABLE MY_TABLE (
    id NUMBER,
    c1 NUMBER,
    ts STRING
);

ACCOUNT_IDENTIFIER:

ACCOUNT_IDENTIFIER에는 조직 내 계정 이름을 사용하는 Format 1 — 예: myorg-account123 — 을 권장합니다. 형식에 대한 자세한 내용은 계정 식별자 를 참고하세요.

설치된 도구:

curl: HTTP 요청용.
jq: JSON 응답 파싱용.
SnowSQL: Snowflake의 명령줄 클라이언트로 명령 실행용.

생성된 JWT:

SnowSQL을 사용해 JWT를 생성하세요.

snowsql --private-key-path rsa_key.p8 --generate-jwt \
  -a  \
  -u MY_USER

주의

JWT를 안전하게 저장하세요. 로그나 스크립트에 노출하지 마세요.

사전 요구 사항 및 설정

다음 단계는 Named Channel REST 수집에 필요한 환경 변수, ingest 호스트, 샘플 행을 구성합니다.

1단계: 환경 변수 설정

Snowflake 계정과 스트리밍 작업에 필요한 환경 변수를 설정하세요. PIPE 변수가 테이블과 연결된 기본 스트리밍 파이프를 대상으로 한다는 점에 유의하세요.

# Paste the JWT token obtained from SnowSQL
export JWT_TOKEN="PASTE_YOUR_JWT_TOKEN_HERE"

# Configure your Snowflake account and resources:
export ACCOUNT="" # For example, ab12345
export USER="MY_USER"
export DB="MY_DATABASE"
export SCHEMA="MY_SCHEMA"
export TABLE="MY_TABLE"

# Replace ACCOUNT with your Account URL Host to form the control plane host:
export CONTROL_HOST="${ACCOUNT}.snowflakecomputing.com"

2단계: ingest 호스트 발견 및 구성

Snowpipe Streaming REST는 두 개의 호스트 이름을 사용합니다: Snowflake 계정 엔드포인트(CONTROL_HOST)와 GET /v2/streaming/hostname이 반환하는 계정 특정 ingest 엔드포인트(INGEST_HOST).

AWS PrivateLink, Azure Private Link, Google Cloud Private Service Connect를 사용한다면 CONTROL_HOST를 SYSTEM$GET_PRIVATELINK_CONFIG 가 반환하는 privatelink-account-url 값의 호스트 이름으로 설정하세요. https://나 후행 슬래시를 포함하지 마세요.

중요

Snowflake 계정 이름에 밑줄이 포함된 경우(예: MY_ACCOUNT), 알려진 문제로 수집 서비스를 호출할 때 내부 오류가 발생할 수 있습니다.

scoped token을 생성하기 전에 INGEST_HOST의 모든 밑줄을 대시로 바꿔야 합니다. 이 변환된 형식(대시 포함)이 이후의 모든 REST API 호출에서 — scoped token 생성 자체를 포함해 — 사용되어야 합니다.

예를 들어 반환된 호스트 이름이 my_account.region.ingest.snowflakecomputing.com이라면 이후 모든 REST API 호출에서 my-account.region.ingest.snowflakecomputing.com으로 바꿔야 합니다.

ingest 호스트는 데이터를 스트리밍하는 엔드포인트입니다. JWT를 사용해 ingest 호스트를 발견하세요.

export INGEST_HOST=$(curl -sS -X GET \
  -H "Authorization: Bearer ***" \
  -H "X-Snowflake-Authorization-Token-Type: KEYPAIR_JWT" \
  "https://${CONTROL_HOST}/v2/streaming/hostname")

echo "Ingest Host: $INGEST_HOST"

ingest 호스트용 프라이빗 DNS 구성

프라이빗 연결을 사용한다면 반환된 ingest 호스트 이름도 Snowflake 프라이빗 엔드포인트를 통해 해석되어야 합니다. ingest 호스트 이름은 Snowflake 계정 호스트 이름과 별개이며 SYSTEM$GET_PRIVATELINK_CONFIG 또는 SYSTEM$ALLOWLIST_PRIVATELINK 에 나타나지 않을 수 있습니다.

정확히 정규화된 INGEST_HOST 값에 대한 프라이빗 DNS 레코드를 만들고 계정 호스트 이름이 사용하는 동일한 기존 Snowflake 프라이빗 엔드포인트로 라우팅하세요.

AWS에서 기존 Snowflake VPC 엔드포인트 리전 DNS 이름을 대상으로 Route 53 프라이빗 호스팅 영역 CNAME(또는 적절한 alias)을 만드세요.
Azure 에서 프라이빗 DNS 구성을 사용해 ingest 호스트 이름을 기존 Snowflake 프라이빗 엔드포인트 IP 주소로 해석하세요.
Google Cloud 에서 ingest 호스트 이름을 기존 Private Service Connect 엔드포인트로 해석하세요.

ingest 호스트 이름에 대해 두 번째 Snowflake 프라이빗 엔드포인트를 만들 필요는 없습니다.

스트리밍 요청을 보내는 동일한 VM, 컨테이너, pod, connector 워커 또는 온프레미스 런타임에서 DNS와 TLS를 검증하세요. TLS SNI (Server Name Indication) 및 HTTP 라우팅을 위해 INGEST_HOST를 요청 호스트 이름으로 유지하세요. 프라이빗 엔드포인트 IP 주소로 바꾸거나 TLS 검증을 비활성화하지 마세요.

ingest 호스트에서 작업을 승인하기 위해 scoped token을 얻으세요.

export SCOPED_TOKEN=$(curl -sS -X POST "https://$CONTROL_HOST/oauth/token" \
  -H 'Content-Type: application/x-www-form-urlencoded' \
  -H "Authorization: Bearer ***" \
  -d "grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer&scope=${INGEST_HOST}")

echo "Scoped Token obtained for ingest host"

3단계: 샘플 행 생성

newline-delimited JSON (NDJSON) 형식으로 배치를 만드세요.

export NOW_TS=$(date -u +"%Y-%m-%dT%H:%M:%SZ")

cat <<EOF > rows.ndjson
{"id":1,"c1":$RANDOM,"ts":"$NOW_TS"}
EOF

Named Channel 열고 사용하기

순서 또는 정확히 한 번 복구가 필요할 때 Named Channel을 사용하세요. 사전 요구 사항 및 설정 의 1~3단계를 완료한 다음 Named Channel 변수를 설정하세요.

export PIPE="MY_TABLE-STREAMING"
export CHANNEL="MY_CHANNEL"

4단계: Named Channel 열기

curl -sS -X PUT \
  -H "Authorization: Bearer ***" \
  -H "Content-Type: application/json" \
  "https://${INGEST_HOST}/v2/streaming/databases/$DB/schemas/$SCHEMA/pipes/$PIPE/channels/$CHANNEL" \
  -d '{"offset_token":"0"}' | tee open_resp.json | jq .

5단계: offset 및 continuation token으로 행 추가

open 작업이 반환한 continuation token과 애플리케이션 관리 소스 offset을 사용하세요.

export CONT_TOKEN=$(jq -r '.next_continuation_token' open_resp.json)
export OFFSET_TOKEN="1"

curl -sS -X POST \
  -H "Authorization: Bearer ***" \
  -H "Content-Type: application/x-ndjson" \
  "https://${INGEST_HOST}/v2/streaming/data/databases/$DB/schemas/$SCHEMA/pipes/$PIPE/channels/$CHANNEL/rows?continuationToken=$CONT_TOKEN&startOffsetToken=$OFFSET_TOKEN&endOffsetToken=$OFFSET_TOKEN" \
  --data-binary @rows.ndjson | tee append_resp.json | jq .

각 append 응답의 next_continuation_token을 다음 append 요청에서 사용하세요.

6단계: 커밋된 진행 상황 검증

curl -sS -X POST \
  -H "Authorization: Bearer ***" \
  -H "Content-Type: application/json" \
  "https://${INGEST_HOST}/v2/streaming/databases/$DB/schemas/$SCHEMA/pipes/$PIPE:bulk-channel-status" \
  -d "{\"channel_names\":[\"$CHANNEL\"]}" | jq ".channel_statuses.\"$CHANNEL\""

소스 offset을 전진시키기 전에 last_committed_offset_token이 1이 될 때까지 기다리세요.

7단계: 데이터 검증

last_committed_offset_token이 전진한 후 대상 테이블을 쿼리하세요.

SELECT * FROM MY_DATABASE.MY_SCHEMA.MY_TABLE ORDER BY id;

(선택) 8단계: 정리

rm -f rows.ndjson open_resp.json append_resp.json
unset JWT_TOKEN SCOPED_TOKEN ACCOUNT USER DB SCHEMA TABLE PIPE CHANNEL CONTROL_HOST INGEST_HOST NOW_TS CONT_TOKEN OFFSET_TOKEN

문제 해결

HTTP 401 (권한 없음): JWT 토큰이 유효하고 만료되지 않았는지 확인하세요. 필요하면 다시 생성하세요.
HTTP 404 (찾을 수 없음): 데이터베이스, 스키마, 테이블 또는 파이프 이름의 철자가 정확하고 Snowflake 계정에 존재하는지 다시 확인하세요.
HTTP 429 (요청이 너무 많음): 재시도 지연에 무작위 변동(jitter)을 가한 지수 백오프로 재시도하세요. 고정된 예약 요청률을 가정하지 마세요.
Ingest 호스트 없음: 제어 플레인 호스트 URL이 정확하고 접근 가능한지 확인하세요.

프라이빗 연결 문제 해결

CONTROL_HOST가 해석되지 않음: privatelink-account-url을 사용했는지, 프라이빗 DNS 영역이 애플리케이션 런타임에 연결되거나 포워딩되는지 확인하세요.
Get Hostname은 성공하지만 INGEST_HOST가 해석되지 않음: 반환된 ingest 호스트 이름을 프라이빗 DNS에 추가하고 기존 Snowflake 프라이빗 엔드포인트를 통해 라우팅하세요.
DNS가 커넥터 밖에서는 작동하지만 내부에서는 실패: 커넥터 컨테이너나 관리형 런타임에서 테스트하고, DNS 변경 후 음수 DNS 응답을 캐시하는 장기 실행 워커를 재시작하세요.
TLS 호스트 이름 불일치: 반환된 ingest 호스트 이름을 URL 호스트 이름으로 유지하세요. 원시 프라이빗 엔드포인트 IP 주소에 연결하거나 TLS SNI 또는 HTTP Host 헤더를 다시 쓰지 마세요.

튜토리얼: REST API로 Named Channel 시작하기