Amazon S3용 Snowpipe 자동화

Amazon S3용 Snowpipe 자동화

이 주제는 S3 버킷에 대한 Amazon SQS(Simple Queue Service) 알림을 사용해 S3의 외부 스테이지에서 Snowpipe 데이터 로드를 자동으로 트리거하는 지침을 제공해요.

Snowflake는 비용, 이벤트 노이즈, 지연 시간을 줄이기 위해 Snowpipe에 지원되는 이벤트만 보낼 것을 권장해요.

출처: Documentation

본문

클라우드 플랫폼 지원

S3 이벤트 메시지를 사용한 자동 Snowpipe 데이터 로드 트리거는 모든 지원되는 클라우드 플랫폼에 호스팅된 Snowflake 계정에서 지원돼요.

네트워크 트래픽

VPS(Virtual Private Snowflake) 및 AWS PrivateLink 고객 참고 사항:

Amazon SQS 알림을 사용한 Snowpipe 자동화는 잘 작동해요. 그러나 VPC(VPS 포함) 안의 AWS 클라우드 스토리지는 자체 메시징 서비스(Amazon SQS, Amazon Simple Notification Service)와 통신할 수 있지만, 이 트래픽은 VPC 외부에 있는 Amazon의 보안 네트워크의 서버 간에 흐르므로 VPC에 의해 보호되지 않아요.

클라우드 스토리지에 대한 보안 액세스 구성

참고 — 데이터 파일을 저장하는 S3 버킷에 대한 보안 액세스를 이미 구성했다면 이 섹션을 건너뛸 수 있어요.

이 섹션은 스토리지 통합을 사용해 Snowflake가 외부(즉, S3) 스테이지에 참조된 Amazon S3 버킷에서 데이터를 읽고 쓸 수 있게 하는 방법을 설명해요. 통합은 시크릿 키나 액세스 토큰 같은 명시적 클라우드 제공업체 자격 증명을 전달할 필요가 없게 하는 이름이 있는 일급(first-class) Snowflake 객체예요. 통합 객체는 AWS IAM(Identity and Access Management) 사용자 ID를 저장해요. 조직의 관리자가 AWS 계정에서 통합 IAM 사용자 권한을 부여해요.

통합은 또한 통합을 사용하는 외부 스테이지를 만들 때 사용자가 지정할 수 있는 위치를 제한하는 버킷(및 선택적 경로)을 나열할 수 있어요.

참고

  • 이 섹션의 지침을 완료하려면 IAM 정책과 역할을 만들고 관리할 AWS 권한이 필요해요. AWS 관리자가 아니라면 관리자에게 이 작업을 수행하도록 요청해요.
  • 현재 스토리지 통합을 사용한 정부 리전의 S3 스토리지 액세스는 같은 정부 리전의 AWS에 호스팅된 Snowflake 계정으로 제한된다는 점에 주의해요. 정부 리전 밖에 호스팅된 계정에서 직접 자격 증명을 사용해 S3 스토리지에 접근하는 것은 지원돼요.

다음 다이어그램은 S3 스테이지의 통합 흐름을 보여줘요.

  • 외부(즉, S3) 스테이지가 정의에서 스토리지 통합 객체를 참조해요.
  • Snowflake는 스토리지 통합을 계정을 위해 만들어진 S3 IAM 사용자와 자동으로 연결해요. Snowflake는 Snowflake 계정의 모든 S3 스토리지 통합이 참조하는 단일 IAM 사용자를 만들어요.
  • 조직의 AWS 관리자가 스테이지 정의에 참조된 버킷에 접근할 수 있도록 IAM 사용자에게 권한을 부여해요. 많은 외부 스테이지 객체가 서로 다른 버킷과 경로를 참조하고 같은 스토리지 통합을 인증에 사용할 수 있다는 점에 주의해요.

사용자가 스테이지에서 데이터를 로드하거나 언로드할 때 Snowflake는 액세스를 허용하거나 거부하기 전에 버킷에서 IAM 사용자에 부여된 권한을 검증해요.

참고 — 이 옵션을 매우 권장해요. 클라우드 스토리지에 접근할 때 IAM 자격 증명을 제공할 필요가 없어져요. 추가 스토리지 액세스 옵션은 Amazon S3에 대한 보안 액세스 구성을 참고해요.

1단계: S3 버킷에 대한 액세스 권한 구성

AWS 액세스 제어 요구 사항

Snowflake가 폴더(및 하위 폴더)의 파일에 접근하려면 S3 버킷과 폴더에 다음 권한이 필요해요.

  • s3:GetBucketLocation
  • s3:GetObject
  • s3:GetObjectVersion
  • s3:ListBucket

모범 사례로, Snowflake가 S3 버킷에 접근하도록 IAM 정책을 만드는 것을 권장해요. 그런 다음 정책을 역할에 연결하고 AWS가 역할에 대해 생성한 보안 자격 증명을 사용해 버킷의 파일에 접근할 수 있어요.

IAM 정책 만들기

다음 단계별 지침은 AWS Management Console에서 Snowflake에 대한 액세스 권한을 구성해 S3 버킷에 접근하는 방법을 설명해요.

  1. AWS Management Console에 로그인해요.
  2. 홈 대시보드에서 IAM을 검색해 선택해요.
  3. 왼쪽 탐색 창에서 Account settings를 선택해요.
  4. Endpoints 목록의 Security Token Service(STS) 아래에서 계정이 위치한 Snowflake 리전을 찾아요. STS status가 비활성(inactive)이면 토글을 Active로 이동해요.
  5. 왼쪽 탐색 창에서 Policies를 선택해요.
  6. Create Policy를 선택해요.
  7. Policy editor에서 JSON을 선택해요.
  8. Snowflake가 S3 버킷과 폴더에 접근할 수 있게 하는 정책 문서를 추가해요.

다음 정책(JSON 형식)은 단일 버킷과 폴더 경로를 사용해 데이터를 로드하거나 언로드하는 데 필요한 권한을 Snowflake에 제공해요.

텍스트를 정책 편집기에 복사해 붙여넣어요.

참고

  • *bucket*과 *prefix*를 실제 버킷 이름과 폴더 경로 접두사로 바꿔야 해요.
  • 정부 리전의 버킷에 대한 Amazon 리소스 이름(ARN)은 arn:aws-us-gov:s3::: 접두사를 가져요.
{
 "Version": "2012-10-17",
 "Statement": [
     {
         "Effect": "Allow",
         "Action": [
           "s3:GetObject",
           "s3:GetObjectVersion"
         ],
         "Resource": "arn:aws:s3:::<bucket>/<prefix>/*"
     },
     {
         "Effect": "Allow",
         "Action": [
             "s3:ListBucket",
             "s3:GetBucketLocation"
         ],
         "Resource": "arn:aws:s3:::<bucket>",
         "Condition": {
             "StringLike": {
                 "s3:prefix": [
                     "<prefix>/*"
                 ]
             }
         }
     }
 ]
}

참고 — "s3:prefix": 조건을 ["*"]으로 설정하면 지정된 버킷의 모든 접두사에 대한 액세스를 부여하고, ["<path>/*"]으로 설정하면 버킷의 해당 경로의 모든 접두사에 대한 액세스를 각각 부여해요. AWS 정책은 다양한 보안 사용 사례를 지원한다는 점에 주의해요.

  1. Next를 선택해요.
  2. Policy name(예: snowflake_access)과 선택적인 Description을 입력해요.
  3. Create policy를 선택해요.

2단계: AWS에서 IAM 역할 만들기

AWS Management Console에서 Snowflake에 대한 액세스 권한을 구성하려면 다음을 수행해요.

  1. IAM(Identity and Access Management) Dashboard의 왼쪽 탐색 창에서 Roles를 선택해요.
  2. Create role을 선택해요.
  3. 신뢰 엔터티 유형으로 AWS account를 선택해요.
  4. Another AWS account를 선택해요.
  5. Account ID 필드에 일시적으로 자신의 AWS 계정 ID를 입력해요. 나중에 신뢰 관계를 수정하고 Snowflake에 액세스를 부여해요.
  6. Require external ID 옵션을 선택해요. 외부 ID는 S3 버킷 같은 AWS 리소스에 대한 액세스를 Snowflake 같은 타사에 부여하는 데 사용돼요. 0000 같은 자리 표시자 ID를 입력해요. 이후 단계에서 IAM 역할의 신뢰 관계를 수정하고 스토리지 통합의 외부 ID를 지정할 거예요.
  7. Next를 선택해요.
  8. (이 주제의) 1단계: S3 버킷에 대한 액세스 권한 구성에서 만든 정책을 선택해요.
  9. Next를 선택해요.
  10. 역할 이름과 설명을 입력한 다음 Create role을 선택해요.

이제 버킷에 대한 IAM 정책을 만들고, IAM 역할을 만들고, 정책을 역할에 연결했어요.

  1. 역할 요약 페이지에서 Role ARN 값을 찾아 기록해요. 다음 단계에서 이 역할을 참조하는 Snowflake 통합을 만들 거예요.

참고 — Snowflake는 60분 만료 시간을 초과할 수 없는 기간 동안 임시 자격 증명을 캐시해요. Snowflake에서 액세스를 취소하면 사용자는 캐시가 만료될 때까지 클라우드 스토리지 위치에서 파일을 나열하고 데이터에 접근할 수 있을 수 있어요.

3단계: Snowflake에서 클라우드 스토리지 통합 만들기

CREATE STORAGE INTEGRATION 명령으로 스토리지 통합을 만들어요. 스토리지 통합은 S3 클라우드 스토리지용으로 생성된 IAM(Identity and Access Management) 사용자와 허용 또는 차단된 스토리지 위치(즉, 버킷)의 선택적 집합을 저장하는 Snowflake 객체예요. 조직의 클라우드 제공업체 관리자가 생성된 사용자에게 스토리지 위치에 대한 권한을 부여해요. 이 옵션은 사용자가 스테이지를 만들거나 데이터를 로드할 때 자격 증명을 제공하지 않아도 되게 해요.

단일 스토리지 통합은 여러 외부(즉, S3) 스테이지를 지원할 수 있어요. 스테이지 정의의 URL은 STORAGE_ALLOWED_LOCATIONS 매개 변수에 지정된 S3 버킷(및 선택적 경로)과 일치해야 해요.

참고 — 계정 관리자(ACCOUNTADMIN 역할을 가진 사용자) 또는 전역 CREATE INTEGRATION 권한이 있는 역할만 이 SQL 명령을 실행할 수 있어요.

CREATE STORAGE INTEGRATION <integration_name>
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = '<iam_role>'
  STORAGE_ALLOWED_LOCATIONS = ('<protocol>://<bucket>/<path>/', '<protocol>://<bucket>/<path>/')
  [ STORAGE_BLOCKED_LOCATIONS = ('<protocol>://<bucket>/<path>/', '<protocol>://<bucket>/<path>/') ]

여기서:

  • integration_name은 새 통합의 이름.
  • iam_role은 (이 주제의) 2단계: AWS에서 IAM 역할 만들기에서 만든 역할의 Amazon 리소스 이름(ARN).
  • protocol은 다음 중 하나.
    • s3 — 중국 밖의 공용 AWS 리전의 S3 스토리지.
    • s3china — 중국의 공용 AWS 리전의 S3 스토리지.
    • s3gov — 정부 리전의 S3 스토리지.
  • bucket은 데이터 파일을 저장하는 S3 버킷의 이름(예: mybucket). 필수 STORAGE_ALLOWED_LOCATIONS 매개 변수와 선택 STORAGE_BLOCKED_LOCATIONS 매개 변수는 각각 이 통합을 참조하는 스테이지를 만들거나 수정할 때 이러한 버킷에 대한 액세스를 제한하거나 차단해요.
  • path는 버킷의 객체에 대한 세분화된 제어를 제공하는 데 사용할 수 있는 선택적 경로.

다음 예제는 계정의 모든 버킷에 대한 접근을 허용하지만 정의된 sensitivedata 폴더에 대한 접근을 차단하는 통합을 만들어요.

이 통합을 사용하는 추가 외부 스테이지는 허용된 버킷과 경로를 참조할 수 있어요.

CREATE STORAGE INTEGRATION s3_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::001234567890:role/myrole'
  STORAGE_ALLOWED_LOCATIONS = ('*')
  STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket1/mypath1/sensitivedata/', 's3://mybucket2/mypath2/sensitivedata/');

참고 — 선택적으로 STORAGE_AWS_EXTERNAL_ID 매개 변수를 사용해 자신의 외부 ID를 지정할 수 있어요. 여러 외부 볼륨 및/또는 스토리지 통합에 걸쳐 같은 외부 ID를 사용하려면 이 옵션을 선택할 수 있어요.

4단계: Snowflake 계정의 AWS IAM 사용자 검색

Snowflake 계정을 위해 자동으로 만들어진 IAM 사용자의 ARN을 검색하려면 DESCRIBE INTEGRATION을 사용해요.

DESC INTEGRATION <integration_name>;

여기서:

예를 들어:

DESC INTEGRATION s3_int;

+---------------------------+---------------+--------------------------------------------------------------------------------+------------------+
| property                  | property_type | property_value                                                                 | property_default |
+---------------------------+---------------+--------------------------------------------------------------------------------+------------------+
| ENABLED                   | Boolean       | true                                                                           | false            |
| STORAGE_ALLOWED_LOCATIONS | List          | s3://mybucket1/mypath1/,s3://mybucket2/mypath2/                                | []               |
| STORAGE_BLOCKED_LOCATIONS | List          | s3://mybucket1/mypath1/sensitivedata/,s3://mybucket2/mypath2/sensitivedata/    | []               |
| STORAGE_AWS_IAM_USER_ARN  | String        | arn:aws:iam::123456789001:user/abc1-b-self1234                                 |                  |
| STORAGE_AWS_ROLE_ARN      | String        | arn:aws:iam::001234567890:role/myrole                                          |                  |
| STORAGE_AWS_EXTERNAL_ID   | String        | MYACCOUNT_SFCRole=2_a123456/s0aBCDEfGHIJklmNoPq=                               |                  |
+---------------------------+---------------+--------------------------------------------------------------------------------+------------------+

다음 속성의 값을 기록해요.

속성 설명
STORAGE_AWS_IAM_USER_ARN Snowflake 계정을 위해 만들어진 AWS IAM 사용자. 예: arn:aws:iam::123456789001:user/abc1-b-self1234. Snowflake는 전체 Snowflake 계정에 대해 단일 IAM 사용자를 프로비저닝해요. 계정의 모든 S3 스토리지 통합이 그 IAM 사용자를 사용해요.
STORAGE_AWS_EXTERNAL_ID Snowflake가 AWS와 신뢰 관계를 수립하는 데 사용하는 외부 ID. 스토리지 통합을 만들 때 외부 ID(STORAGE_AWS_EXTERNAL_ID)를 지정하지 않았다면 Snowflake가 사용할 ID를 생성해요.

다음 섹션에서 이 값을 제공해요.

5단계: 버킷 객체에 접근하도록 IAM 사용자에게 권한 부여

다음 단계별 지침은 AWS Management Console에서 Snowflake에 대한 IAM 액세스 권한을 구성해 S3 버킷을 사용해 데이터를 로드하고 언로드하는 방법을 설명해요.

  1. AWS Management Console에 로그인해요.
  2. IAM을 선택해요.
  3. 왼쪽 탐색 창에서 Roles를 선택해요.
  4. (이 주제의) 2단계: AWS에서 IAM 역할 만들기에서 만든 역할을 선택해요.
  5. Trust relationships 탭을 선택해요.
  6. Edit trust policy를 선택해요.
  7. (이 주제의) 4단계: Snowflake 계정의 AWS IAM 사용자 검색에서 기록한 DESC STORAGE INTEGRATION 출력 값으로 정책 문서를 수정해요.

IAM 역할용 정책 문서

{
  "Version": "2012-10-17",
  "Statement": [
 {
   "Sid": "",
   "Effect": "Allow",
   "Principal": {
     "AWS": "<snowflake_user_arn>"
   },
   "Action": "sts:AssumeRole",
   "Condition": {
     "StringEquals": {
       "sts:ExternalId": "<snowflake_external_id>"
     }
   }
 }
  ]
}

여기서:

  • snowflake_user_arn은 기록한 STORAGE_AWS_IAM_USER_ARN 값.
  • snowflake_external_id는 기록한 STORAGE_AWS_EXTERNAL_ID 값.

이 예제에서 snowflake_external_id 값은 MYACCOUNT_SFCRole=2_a123456/s0aBCDEfGHIJklmNoPq=이에요.

참고 — 보안상의 이유로 외부 ID를 지정하지 않고 새 스토리지 통합을 만들거나(CREATE OR REPLACE STORAGE INTEGRATION 문법으로 기존 통합을 다시 만들면) 새 통합은 다른 외부 ID를 가지며, 신뢰 정책을 업데이트하지 않으면 신뢰 관계를 해결할 수 없어요.

  1. 변경 사항을 저장하려면 Update policy를 선택해요.

참고 — Snowflake는 60분 만료 시간을 초과할 수 없는 기간 동안 임시 자격 증명을 캐시해요. Snowflake에서 액세스를 취소하면 사용자는 캐시가 만료될 때까지 클라우드 스토리지 위치에서 파일을 나열하고 데이터를 로드할 수 있을 수 있어요.

참고 — SYSTEM$VALIDATE_STORAGE_INTEGRATION 함수를 사용해 스토리지 통합의 구성을 검증할 수 있어요.

올바른 옵션 결정

진행하기 전에 데이터 파일이 있는 S3 버킷의 대상 경로(또는 AWS 용어의 "접두사")에 대한 S3 이벤트 알림이 존재하는지 결정해요. AWS 규칙은 같은 경로에 대해 충돌하는 알림을 만드는 것을 금지해요.

Amazon SQS를 사용해 Snowpipe를 자동화하는 다음 옵션이 지원돼요.

  • 옵션 1. 새 S3 이벤트 알림: S3 버킷의 대상 경로에 대한 이벤트 알림을 만들어요. 이벤트 알림은 파일이 로드할 준비가 되었을 때 SQS 큐를 통해 Snowpipe에 알려줘요.

중요 — S3 버킷에 대해 충돌하는 이벤트 알림이 존재한다면 대신 옵션 2를 사용해요.

  • 옵션 2. 기존 이벤트 알림: Amazon Simple Notification Service(SNS)를 브로드캐스터로 구성해 주어진 경로에 대한 알림을 여러 엔드포인트("구독자"(예: SQS 큐 또는 AWS Lambda 워크로드)와 Snowpipe 자동화용 Snowflake SQS 큐를 포함)와 공유해요. SNS가 게시한 S3 이벤트 알림은 파일이 로드할 준비가 되었을 때 SQS 큐를 통해 Snowpipe에 알려줘요.

참고 — 스테이지, 파이프 및 로드 기록 복제를 사용할 계획이라면 이 옵션을 권장해요. 복제 또는 장애 조치 그룹을 만든 후 옵션 1에서 옵션 2로 마이그레이션할 수도 있어요. 자세한 내용은 Amazon Simple Notification Service(SNS)로 마이그레이션을 참고해요.

  • 옵션 3. Snowpipe 자동화를 위한 Amazon EventBridge 설정: 옵션 2와 유사하게 S3 버킷에 대해 Amazon EventBridge를 활성화하고 SNS 토픽에 알림을 보내는 규칙을 만들 수 있어요.

옵션 1: Snowpipe를 자동화하기 위한 새 S3 이벤트 알림 만들기

이 섹션은 Amazon SQS(Simple Queue Service) 알림을 S3 버킷에 사용해 Snowpipe 데이터 로드를 자동으로 트리거하는 가장 일반적인 옵션을 설명해요. 단계는 데이터 파일이 저장된 S3 버킷의 대상 경로(또는 AWS 용어의 "접두사")에 대한 이벤트 알림을 만드는 방법을 설명해요.

중요 — S3 버킷에 대해 충돌하는 이벤트 알림이 존재한다면 (이 주제의) 옵션 2: SQS 알림을 사용해 Snowpipe를 자동화하도록 Amazon SNS 구성을 대신 사용해요. AWS 규칙은 같은 대상 경로에 대해 충돌하는 알림을 만드는 것을 금지해요.

Snowpipe 자동 수집 프로세스 흐름:

  1. 데이터 파일이 스테이지에 로드됐어요.
  2. S3 이벤트 알림이 SQS 큐를 통해 파일을 로드할 준비가 되었음을 Snowpipe에 알려줘요. Snowpipe는 파일을 큐에 복사해요.
  3. Snowflake가 제공하는 가상 웨어하우스가 지정된 파이프에 정의된 매개 변수에 따라 큐에 있는 파일의 데이터를 대상 테이블에 로드해요.

참고 — 이 주제의 지침은 데이터가 로드될 Snowflake 데이터베이스에 이미 대상 테이블이 있다고 가정해요.

1단계: (필요하면) 스테이지 만들기

CREATE STAGE 명령을 사용해 S3 버킷을 참조하는 외부 스테이지를 만들어요. Snowpipe는 스테이지에서 데이터 파일을 가져와 대상 테이블에 로드하기 전에 임시로 큐에 넣어요. 또는 기존 외부 스테이지를 사용할 수 있어요.

참고

  • 클라우드 스토리지 위치에 대한 보안 액세스를 구성하려면 (이 주제의) 클라우드 스토리지에 대한 보안 액세스 구성을 참고해요.
  • CREATE STAGE 문에서 스토리지 통합을 참조하려면 역할이 스토리지 통합 객체에 USAGE 권한이 있어야 해요.

다음 예제는 사용자 세션의 활성 스키마에 mystage라는 스테이지를 만들어요. 클라우드 스토리지 URL에는 경로 files가 포함돼요. 스테이지는 my_storage_int라는 스토리지 통합을 참조해요.

USE SCHEMA snowpipe_db.public;

CREATE STAGE mystage
  URL = 's3://mybucket/load/files'
  STORAGE_INTEGRATION = my_storage_int;

2단계: 자동 수집이 활성화된 파이프 만들기

CREATE PIPE 명령을 사용해 파이프를 만들어요. 파이프는 Snowpipe가 수집 큐에서 대상 테이블로 데이터를 로드하는 데 사용하는 COPY INTO <table> 문을 정의해요.

다음 예제는 사용자 세션의 활성 스키마에 mypipe라는 파이프를 만들어요. 이 파이프는 mystage 스테이지에 스테이징된 파일의 데이터를 mytable 테이블에 로드해요.

CREATE PIPE snowpipe_db.public.mypipe
  AUTO_INGEST = TRUE
  AS
    COPY INTO snowpipe_db.public.mytable
      FROM @snowpipe_db.public.mystage
      FILE_FORMAT = (type = 'JSON');

AUTO_INGEST = TRUE 매개 변수는 로드할 새 데이터가 준비되었을 때 S3 버킷에서 SQS 큐로 전송된 이벤트 알림을 읽도록 지정해요.

중요 — 파이프 정의의 스테이지 참조를 기존 파이프와 비교해요. 같은 S3 버킷에 대한 디렉터리 경로가 겹치지 않는지 확인해요. 그렇지 않으면 여러 파이프가 같은 데이터 파일 집합을 하나 이상의 대상 테이블에 여러 번 로드할 수 있어요. 예를 들어 여러 스테이지가 s3://mybucket/path1과 s3://mybucket/path1/path2 같이 서로 다른 세분화 수준으로 같은 S3 버킷을 참조하면 이런 일이 발생할 수 있어요. 이 사용 사례에서 파일이 s3://mybucket/path1/path2에 스테이징되면 두 스테이지의 파이프 모두 파일 복사본을 로드하게 돼요.

이것은 사용자가 로드할 이름이 있는 파일 집합을 REST API에 제출해 파일을 큐에 넣어야 하는 수동 Snowpipe 설정(자동 수집 비활성화)과 다릅니다. 자동 수집이 활성화되면 각 파이프는 S3 이벤트 알림에서 생성된 파일 목록을 받아요. 데이터 중복을 피하기 위한 추가 주의가 필요해요.

3단계: 보안 구성

Snowpipe를 사용해 연속 데이터 로드를 실행할 각 사용자에 대해 데이터 로드(즉, 대상 데이터베이스, 스키마, 테이블), 스테이지 객체, 파이프에 대한 객체에 충분한 액세스 제어 권한을 부여해요.

참고 — "최소 권한"의 일반 원칙을 따르기 위해, 파이프를 사용해 파일을 수집하는 데 사용할 별도의 사용자와 역할을 만드는 것을 권장해요. 사용자는 이 역할을 기본 역할로 만들어야 해요.

Snowpipe를 사용하려면 다음 권한이 있는 역할이 필요해요.

객체 권한 비고
이름이 있는 파이프 OWNERSHIP
이름이 있는 스테이지 USAGE, READ
이름이 있는 파일 형식 USAGE 선택 사항. (필요하면) 작성한 스테이지가 이름이 있는 파일 형식을 참조하는 경우에만 필요.
대상 데이터베이스 USAGE
대상 스키마 USAGE
대상 테이블 INSERT, SELECT

GRANT <privileges> … TO ROLE 명령을 사용해 권한을 역할에 부여해요.

참고 — 보안 관리자(즉, SECURITYADMIN 역할을 가진 사용자) 또는 그 이상, 또는 계정에 CREATE ROLE 권한과 전역 MANAGE GRANTS 권한을 모두 가진 다른 역할만 역할을 만들고 권한을 부여할 수 있어요.

예를 들어 snowpipe_db.public 데이터베이스 객체 집합과 mypipe라는 파이프에 접근할 수 있는 snowpipe_role이라는 역할을 만든 다음 그 역할을 사용자에게 부여해요.

-- Create a role to contain the Snowpipe privileges
USE ROLE SECURITYADMIN;

CREATE OR REPLACE ROLE snowpipe_role;

-- Grant the required privileges on the database objects
GRANT USAGE ON DATABASE snowpipe_db TO ROLE snowpipe_role;

GRANT USAGE ON SCHEMA snowpipe_db.public TO ROLE snowpipe_role;

GRANT INSERT, SELECT ON snowpipe_db.public.mytable TO ROLE snowpipe_role;

GRANT USAGE ON STAGE snowpipe_db.public.mystage TO ROLE snowpipe_role;

-- Pause the pipe for OWNERSHIP transfer
ALTER PIPE mypipe SET PIPE_EXECUTION_PAUSED = TRUE;

-- Grant the OWNERSHIP privilege on the pipe object
GRANT OWNERSHIP ON PIPE snowpipe_db.public.mypipe TO ROLE snowpipe_role;

-- Grant the role to a user
GRANT ROLE snowpipe_role TO USER jsmith;

-- Set the role as the default role for the user
ALTER USER jsmith SET DEFAULT_ROLE = snowpipe_role;

-- Resume the pipe
ALTER PIPE mypipe SET PIPE_EXECUTION_PAUSED = FALSE;

4단계: 이벤트 알림 구성

로드할 새 데이터가 있을 때 Snowpipe에 알리도록 S3 버킷에 대한 이벤트 알림을 구성해요. 자동 수집 기능은 SQS 큐에 의존해 S3에서 Snowpipe로 이벤트 알림을 전달해요.

사용 편의를 위해 Snowpipe SQS 큐는 Snowflake가 만들고 관리해요. SHOW PIPES 명령 출력은 SQS 큐의 Amazon 리소스 이름(ARN)을 표시해요.

  1. SHOW PIPES 명령을 실행해요.
SHOW PIPES;

notification_channel 컬럼에서 스테이지의 SQS 큐 ARN을 기록해요. ARN을 편리한 위치에 복사해요.

참고 — AWS 지침에 따라 Snowflake는 AWS S3 리전당 최대 하나의 SQS 큐를 지정해요. SQS 큐는 같은 AWS 계정의 같은 리전에 있는 여러 버킷 간에 공유될 수 있어요. SQS 큐는 S3 버킷의 외부 스테이지를 대상 테이블에 연결하는 모든 파이프에 대한 알림을 조정해요. 데이터 파일이 버킷에 업로드되면 스테이지 디렉터리 경로와 일치하는 모든 파이프가 파일을 해당 대상 테이블에 일회성 로드해요.

  1. Amazon S3 콘솔에 로그인해요.
  2. Amazon S3 문서에 제공된 지침을 사용해 S3 버킷에 대한 이벤트 알림을 구성해요. 필드를 다음과 같이 완료해요.
    • Name: 이벤트 알림의 이름(예: Auto-ingest Snowflake).
    • Events: ObjectCreate (All) 옵션을 선택해요.
    • Send to: 드롭다운 목록에서 SQS Queue를 선택해요.
    • SQS: 드롭다운 목록에서 Add SQS queue ARN을 선택해요.
    • SQS queue ARN: SHOW PIPES 출력의 SQS 큐 이름을 붙여넣어요.

참고 — 이 지침은 전체 S3 버킷의 활동을 모니터링하는 단일 이벤트 알림을 만들어요. 이것이 가장 간단한 접근 방식이에요. 이 알림은 S3 버킷 디렉터리에서 더 세분화된 수준으로 구성된 모든 파이프를 처리해요. Snowpipe는 파이프 정의에 지정된 데이터 파일만 로드해요. 그러나 파이프 정의 밖의 활동에 대한 많은 알림 볼륨은 Snowpipe가 알림을 필터링하고 조치를 취하는 속도에 부정적인 영향을 줄 수 있다는 점에 유의해요.

또는 위 단계에서 하나 이상의 경로 및/또는 파일 확장자(또는 AWS 용어의 접두사 및 접미사)를 구성해 이벤트 활동을 필터링해요. 지침은 관련 AWS 문서 토픽의 객체 키 이름 필터링 정보를 참고해요. 알림이 모니터링할 각 추가 경로 또는 파일 확장자에 대해 이 단계를 반복해요.

AWS가 S3 버킷당 이러한 알림 큐 구성의 수를 최대 100개로 제한한다는 점에 주의해요.

또한 AWS는 같은 S3 버킷에 대해 (이벤트 알림 간에) 겹치는 큐 구성을 허용하지 않는다는 점에 주의해요. 예를 들어 기존 알림이 s3://mybucket/load/path1에 대해 구성되어 있다면 s3://mybucket/load 같은 더 높은 수준에서 다른 알림을 만들 수 없으며, 그 반대도 마찬가지예요.

자동 수집 Snowpipe가 이제 구성됐어요!

새 데이터 파일이 S3 버킷에 추가될 때 이벤트 알림이 Snowpipe에 파이프에 정의된 대상 테이블로 로드하라고 알려줘요.

5단계: 이력 파일 로드

SQS 알림이 구성되기 전에 외부 스테이지에 존재했던 데이터 파일의 백로그를 로드하려면 이력 로드를 참고해요.

6단계: 스테이징된 파일 삭제

데이터를 성공적으로 로드하고 더 이상 파일이 필요하지 않으면 스테이징된 파일을 삭제해요. 지침은 Snowpipe가 데이터를 로드한 후 스테이징된 파일 삭제를 참고해요.

옵션 2: SQS 알림을 사용해 Snowpipe를 자동화하도록 Amazon SNS 구성

이 섹션은 Amazon SQS(Simple Queue Service) 알림을 S3 버킷에 사용해 Snowpipe 데이터 로드를 자동으로 트리거하는 방법을 설명해요. 단계는 Amazon Simple Notification Service(SNS)를 브로드캐스터로 구성해 S3 버킷에 대한 이벤트 알림을 여러 구독자(예: SQS 큐 또는 AWS Lambda 워크로드)와 Snowpipe 자동화용 Snowflake SQS 큐에 게시하는 방법을 설명해요.

참고 — 이 지침은 데이터 파일이 있는 S3 버킷의 대상 경로에 대한 이벤트 알림이 존재한다고 가정해요. 이벤트 알림이 없다면 (이 주제의) 옵션 1: Snowpipe를 자동화하기 위한 새 S3 이벤트 알림 만들기를 따르거나, S3 버킷에 대한 이벤트 알림을 만든 후 이 주제의 지침을 진행해요. 자세한 내용은 Amazon S3 문서를 참고해요.

Amazon SNS를 사용한 Snowpipe 자동 수집 프로세스 흐름:

  1. 데이터 파일이 스테이지에 로드됐어요.
  2. SNS가 게시한 S3 이벤트 알림이 SQS 큐를 통해 파일을 로드할 준비가 되었음을 Snowpipe에 알려줘요. Snowpipe는 파일을 큐에 복사해요.
  3. Snowflake가 제공하는 가상 웨어하우스가 지정된 파이프에 정의된 매개 변수에 따라 큐에 있는 파일의 데이터를 대상 테이블에 로드해요.

참고 — 지침은 데이터가 로드될 Snowflake 데이터베이스에 이미 대상 테이블이 있다고 가정해요.

Snowpipe 자동 수집은 AWS KMS로 암호화된 SNS 토픽을 지원해요. 자세한 내용은 미사용 암호화(Encryption at rest)를 참고해요.

전제 조건: Amazon SNS 토픽 및 구독 만들기

  • S3 버킷의 Snowflake 스테이지 위치에 대한 모든 메시지를 처리할 SNS 토픽을 AWS 계정에 만들어요.
  • S3 이벤트 알림의 대상 대상(예: 다른 SQS 큐 또는 AWS Lambda 워크로드)을 이 토픽에 구독해요. SNS는 버킷에 대한 이벤트 알림을 토픽의 모든 구독자에게 게시해요.

지침은 SNS 문서를 참고해요.

1단계: Snowflake SQS 큐를 SNS 토픽에 구독

  1. AWS Management Console에 로그인해요.
  2. 홈 대시보드에서 Simple Notification Service(SNS)를 선택해요.
  3. 왼쪽 탐색 창에서 Topics를 선택해요.
  4. S3 버킷에 대한 토픽을 찾아요. 토픽 ARN을 기록해요.
  5. Snowflake 클라이언트를 사용해 SNS 토픽 ARN으로 SYSTEM$GET_AWS_SNS_IAM_POLICY 시스템 함수를 조회해요.
select system$get_aws_sns_iam_policy('<sns_topic_arn>');

이 함수는 Snowflake SQS 큐가 SNS 토픽에 구독할 권한을 부여하는 IAM 정책을 반환해요.

예를 들어:

select system$get_aws_sns_iam_policy('arn:aws:sns:us-west-2:001234567890:s3_mybucket');

+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| SYSTEM$GET_AWS_SNS_IAM_POLICY('ARN:AWS:SNS:US-WEST-2:001234567890:S3_MYBUCKET')                                                                                                                                                                   |
+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| {"Version":"2012-10-17","Statement":[{"Sid":"1","Effect":"Allow","Principal":{"AWS":"arn:aws:iam::123456789001:user/vj4g-a-abcd1234"},"Action":["sns:Subscribe"],"Resource":["arn:aws:sns:us-west-2:001234567890:s3_mybucket"]}]}                 |
+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
  1. AWS Management Console로 돌아가요. 왼쪽 탐색 창에서 Topics를 선택해요.
  2. S3 버킷에 대한 토픽을 선택하고 Edit 버튼을 클릭해요. Edit 페이지가 열려요.
  3. Access policy - Optional을 클릭해 페이지의 이 영역을 펼쳐요.
  4. SYSTEM$GET_AWS_SNS_IAM_POLICY 함수 결과의 IAM 정책 추가를 JSON 문서에 병합해요.

예를 들어:

원래 IAM 정책(축약):

{
  "Version":"2008-10-17",
  "Id":"__default_policy_ID",
  "Statement":[
  {
     "Sid":"__default_statement_ID",
     "Effect":"Allow",
     "Principal":{
        "AWS":"*"
     }
     ..
  }
]
 }

병합된 IAM 정책:

{
  "Version":"2008-10-17",
  "Id":"__default_policy_ID",
  "Statement":[
  {
     "Sid":"__default_statement_ID",
     "Effect":"Allow",
     "Principal":{
        "AWS":"*"
     }
     ..
  },
  {
     "Sid":"1",
     "Effect":"Allow",
     "Principal":{
       "AWS":"arn:aws:iam::123456789001:user/vj4g-a-abcd1234"
      },
      "Action":[
        "sns:Subscribe"
      ],
      "Resource":[
        "arn:aws:sns:us-west-2:001234567890:s3_mybucket"
      ]
  }
]
 }
  1. S3가 버킷에 대한 이벤트 알림을 SNS 토픽에 게시할 수 있도록 추가 정책 부여를 추가해요.

예를 들어(이 지침 전체에서 사용된 SNS 토픽 ARN과 S3 버킷 사용):

{
 "Sid":"s3-event-notifier",
 "Effect":"Allow",
 "Principal":{
    "Service":"s3.amazonaws.com"
 },
 "Action":"SNS:Publish",
 "Resource":"arn:aws:sns:us-west-2:001234567890:s3_mybucket",
 "Condition":{
    "ArnLike":{
       "aws:SourceArn":"arn:aws:s3:*:*:s3_mybucket"
    }
 }
 }

병합된 IAM 정책:

{
  "Version":"2008-10-17",
  "Id":"__default_policy_ID",
  "Statement":[
  {
     "Sid":"__default_statement_ID",
     "Effect":"Allow",
     "Principal":{
        "AWS":"*"
     }
     ..
  },
  {
     "Sid":"1",
     "Effect":"Allow",
     "Principal":{
       "AWS":"arn:aws:iam::123456789001:user/vj4g-a-abcd1234"
      },
      "Action":[
        "sns:Subscribe"
      ],
      "Resource":[
        "arn:aws:sns:us-west-2:001234567890:s3_mybucket"
      ]
  },
  {
     "Sid":"s3-event-notifier",
     "Effect":"Allow",
     "Principal":{
        "Service":"s3.amazonaws.com"
     },
     "Action":"SNS:Publish",
     "Resource":"arn:aws:sns:us-west-2:001234567890:s3_mybucket",
     "Condition":{
        "ArnLike":{
           "aws:SourceArn":"arn:aws:s3:*:*:s3_mybucket"
        }
     }
   }
]
 }
  1. Save changes를 클릭해요.

2단계: (필요하면) 스테이지 만들기

CREATE STAGE 명령을 사용해 S3 버킷을 참조하는 외부 스테이지를 만들어요. Snowpipe는 스테이지에서 데이터 파일을 가져와 대상 테이블에 로드하기 전에 임시로 큐에 넣어요.

또는 기존 외부 스테이지를 사용할 수 있어요.

참고 — 클라우드 스토리지 위치에 대한 보안 액세스를 구성하려면 (이 주제의) 클라우드 스토리지에 대한 보안 액세스 구성을 참고해요.

다음 예제는 사용자 세션의 활성 스키마에 mystage라는 스테이지를 만들어요. 클라우드 스토리지 URL에는 경로 files가 포함돼요. 스테이지는 my_storage_int라는 스토리지 통합을 참조해요.

CREATE STAGE mystage
  URL = 's3://mybucket/load/files'
  STORAGE_INTEGRATION = my_storage_int;

3단계: 자동 수집이 활성화된 파이프 만들기

CREATE PIPE 명령을 사용해 파이프를 만들어요. 파이프는 Snowpipe가 수집 큐에서 대상 테이블로 데이터를 로드하는 데 사용하는 COPY INTO <table> 문을 정의해요. COPY 문에서 전제 조건: Amazon SNS 토픽 및 구독 만들기의 SNS 토픽 ARN을 식별해요.

다음 예제는 사용자 세션의 활성 스키마에 mypipe라는 파이프를 만들어요. 이 파이프는 mystage 스테이지에 스테이징된 파일의 데이터를 mytable 테이블에 로드해요.

CREATE PIPE snowpipe_db.public.mypipe
  AUTO_INGEST = TRUE
  AWS_SNS_TOPIC='<sns_topic_arn>'
  AS
    COPY INTO snowpipe_db.public.mytable
      FROM @snowpipe_db.public.mystage
      FILE_FORMAT = (type = 'JSON');

여기서:

  • AUTO_INGEST = TRUE — 로드할 새 데이터가 준비되었을 때 S3 버킷에서 SQS 큐로 전송된 이벤트 알림을 읽도록 지정.
  • AWS_SNS_TOPIC = '<sns_topic_arn>' — S3 버킷의 SNS 토픽 ARN을 지정(현재 예제에서는 arn:aws:sns:us-west-2:001234567890:s3_mybucket). CREATE PIPE 문은 Snowflake SQS 큐를 지정된 SNS 토픽에 구독해요. 파이프는 SNS 토픽을 통해 이벤트 알림으로 트리거된 수집 큐에만 파일을 복사한다는 점에 주의해요.

파이프에서 두 매개 변수 중 하나를 제거하려면 현재 CREATE OR REPLACE PIPE 문법을 사용해 파이프를 다시 만들어야 해요.

중요 — COPY INTO <table> 문의 스토리지 위치 참조가 계정의 기존 파이프의 참조와 겹치지 않는지 확인해요. 그렇지 않으면 여러 파이프가 같은 데이터 파일 집합을 대상 테이블에 로드할 수 있어요. 예를 들어 여러 파이프 정의가 <storage_location>/path1/과 <storage_location>/path1/path2/ 같은 서로 다른 세분화 수준으로 같은 스토리지 위치를 참조하면 이 상황이 발생할 수 있어요. 이 예제에서 파일이 <storage_location>/path1/path2/에 스테이징되면 두 파이프 모두 파일 복사본을 로드하게 돼요.

SHOW PIPES를 실행하거나 Account Usage의 PIPES 뷰 또는 정보 스키마의 PIPES 뷰를 조회해 계정의 모든 파이프 정의에서 COPY INTO <table> 문을 확인할 수 있어요.

4단계: 보안 구성

Snowpipe를 사용해 연속 데이터 로드를 실행할 각 사용자에 대해 데이터 로드(즉, 대상 데이터베이스, 스키마, 테이블), 스테이지 객체, 파이프에 대한 객체에 충분한 액세스 제어 권한을 부여해요.

참고 — "최소 권한"의 일반 원칙을 따르기 위해, 파이프를 사용해 파일을 수집하는 데 사용할 별도의 사용자와 역할을 만드는 것을 권장해요. 사용자는 이 역할을 기본 역할로 만들어야 해요.

Snowpipe를 사용하려면 다음 권한이 있는 역할이 필요해요.

객체 권한 비고
이름이 있는 파이프 OWNERSHIP
이름이 있는 스토리지 통합 USAGE (필요하면) 2단계에서 만든 스테이지가 스토리지 통합을 참조하는 경우에 필요.
이름이 있는 스테이지 USAGE, READ
이름이 있는 파일 형식 USAGE 선택 사항. (필요하면) 2단계에서 만든 스테이지가 이름이 있는 파일 형식을 참조하는 경우에만 필요.
대상 데이터베이스 USAGE
대상 스키마 USAGE
대상 테이블 INSERT, SELECT

GRANT <privileges> … TO ROLE 명령을 사용해 권한을 역할에 부여해요.

참고 — 보안 관리자(즉, SECURITYADMIN 역할을 가진 사용자) 또는 그 이상만 역할을 만들 수 있어요.

예를 들어 snowpipe_db.public 데이터베이스 객체 집합과 mypipe라는 파이프에 접근할 수 있는 snowpipe_role이라는 역할을 만든 다음 그 역할을 사용자에게 부여해요.

-- Create a role to contain the Snowpipe privileges
USE ROLE SECURITYADMIN;

CREATE OR REPLACE ROLE snowpipe_role;

-- Grant the required privileges on the database objects
GRANT USAGE ON DATABASE snowpipe_db TO ROLE snowpipe_role;

GRANT USAGE ON SCHEMA snowpipe_db.public TO ROLE snowpipe_role;

GRANT INSERT, SELECT ON snowpipe_db.public.mytable TO ROLE snowpipe_role;

GRANT USAGE, READ ON STAGE snowpipe_db.public.mystage TO ROLE snowpipe_role;

-- Pause the pipe for OWNERSHIP transfer
ALTER PIPE mypipe SET PIPE_EXECUTION_PAUSED = TRUE;

-- Grant the OWNERSHIP privilege on the pipe object
GRANT OWNERSHIP ON PIPE snowpipe_db.public.mypipe TO ROLE snowpipe_role;

-- Grant the role to a user
GRANT ROLE snowpipe_role TO USER jsmith;

-- Set the role as the default role for the user
ALTER USER jsmith SET DEFAULT_ROLE = snowpipe_role;

-- Resume the pipe
ALTER PIPE mypipe SET PIPE_EXECUTION_PAUSED = FALSE;

자동 수집 Snowpipe가 이제 구성됐어요!

새 데이터 파일이 S3 버킷에 추가될 때 이벤트 알림이 Snowpipe에 파이프에 정의된 대상 테이블로 로드하라고 알려줘요.

5단계: 이력 파일 로드

SQS 알림이 구성되기 전에 외부 스테이지에 존재했던 데이터 파일의 백로그를 로드하려면 이력 로드를 참고해요.

6단계: 스테이징된 파일 삭제

데이터를 성공적으로 로드하고 더 이상 파일이 필요하지 않으면 스테이징된 파일을 삭제해요. 지침은 Snowpipe가 데이터를 로드한 후 스테이징된 파일 삭제를 참고해요.

옵션 3: Snowpipe를 자동화하도록 Amazon EventBridge 설정

옵션 2와 유사하게 Amazon EventBridge를 설정해 Snowpipe를 자동화할 수도 있어요.

1단계: Amazon SNS 토픽 만들기

(이 주제의) 전제 조건: Amazon SNS 토픽 및 구독 만들기를 따르세요.

2단계: S3 버킷을 구독하고 SNS 토픽에 알림을 보내는 EventBridge 규칙 만들기

3단계: SQS 알림을 사용해 Snowpipe를 자동화하도록 Amazon SNS 구성

(이 주제의) 옵션 2: SQS 알림을 사용해 Snowpipe를 자동화하도록 Amazon SNS 구성을 따르세요.

SYSTEM$PIPE_STATUS 출력

SYSTEM$PIPE_STATUS 함수는 파이프의 현재 상태에 대한 JSON 표현을 검색해요.

AUTO_INGEST가 TRUE로 설정된 파이프의 경우 이 함수는 (현재 파이프 상태에 해당된다면) 다음 이름/값 쌍을 포함하는 JSON 객체를 반환해요.

{"executionState":"<value>","oldestFileTimestamp":<value>,"pendingFileCount":<value>,"notificationChannelName":"<value>","numOutstandingMessagesOnChannel":<value>,"lastReceivedMessageTimestamp":"<value>","lastForwardedMessageTimestamp":"<value>","error":<value>,"fault":<value>}

출력 값에 대한 설명은 SQL 함수의 참조 주제를 참고해요.

더 알아보기