스테이지, 파이프 및 로드 기록 복제¶

요구 사항¶

데이터 파이프라인 오브젝트를 복제하려면 먼저 ETL 복제를 활성화해야 합니다. ETL 복제 지원은 2024_01 BCR 번들 의 일부입니다. 다음 문을 실행하여 2024_01 BCR 번들을 활성화 할 수 있습니다.

SELECT SYSTEM$ENABLE_BEHAVIOR_CHANGE_BUNDLE('2024_01');

저장소 통합을 사용하는 외부 스테이지를 복제하려면 STORAGE INTEGRATIONS 를 복제하도록 복제 그룹 또는 장애 조치 그룹도 구성해야 합니다. 그렇지 않으면 연결된 저장소 통합 없이 외부 스테이지가 복제됩니다.

ALTER REPLICATION GROUP 또는 ALTER FAILOVER GROUP 문을 사용하여 기존 그룹에 대해 이러한 속성을 수정할 수 있습니다.

예:

ALTER FAILOVER GROUP my_failover_group SET
  OBJECT_TYPES = ROLES, INTEGRATIONS
  ALLOWED_INTEGRATION_TYPES = API INTEGRATIONS, STORAGE INTEGRATIONS;

복제 및 스테이지¶

이 섹션에서는 다양한 스테이지 유형에 대해 Snowflake가 지원하는 복제 기능의 현재 수준에 대해 설명합니다.

복제 및 파이프¶

이 섹션에서는 다양한 파이프 유형에 대해 지원되는 복제 기능의 현재 수준에 대해 설명합니다.

Snowflake는 다음에 대한 복제를 지원합니다.

• 외부 스테이지에서 데이터를 로드하는 자동 수집 및 REST 엔드포인트 파이프를 포함한 파이프 오브젝트.

• 파이프 수준 매개 변수.

• 파이프 오브젝트에 대한 권한 부여.

예 1: 명명된 내부 스테이지 복제하기¶

이 예에서는 내부 스테이지에 대한 복제의 작동 방식을 보여줍니다. 특히 이 예에서는 디렉터리 테이블이 복제 전후의 테이블 메타데이터에 대한 단일 정보 소스가 되는 방식을 보여줍니다.

이 예의 첫 번째 부분에서는 원본 계정에서 다음 작업을 완료합니다.

1. 스테이지의 파일을 복제하기 위해 활성화된 디렉터리 테이블로 my_int_stage 라는 내부 스테이지를 생성합니다. 그런 다음 my_table 이라는 테이블의 데이터를 스테이지의 파일에 복사합니다.

   참고

   이 예에서는 file1 및 file2 를 스테이지에 로드한 후 디렉터리 테이블을 새로 고쳐 테이블 메타데이터를 디렉터리 테이블의 스테이지 정의에 있는 최신 파일 세트와 동기화합니다. 하지만 file3 을 로드한 후에는 새로 고침 작업이 발생하지 않습니다.

   CREATE OR REPLACE STAGE my_stage
     DIRECTORY = (ENABLE = TRUE);
   
   COPY INTO @my_stage/folder1/file1 from my_table;
   COPY INTO @my_stage/folder2/file2 from my_table;
   ALTER STAGE my_stage REFRESH;
   
   COPY INTO @my_stage/folder3/file3 from my_table;
   

   Copy

2. 장애 조치 그룹 만들기:

   CREATE FAILOVER GROUP my_stage_failover_group
     OBJECT_TYPES = DATABASES
     ALLOWED_DATABASES = my_database_1
     ALLOWED_ACCOUNTS = myorg.my_account_2;
   

   Copy

이 예의 두 번째 부분에서는 대상 계정에서 복제 및 장애 조치 프로세스를 완료합니다.

1. 장애 조치 그룹을 원본 계정에 있는 장애 조치 그룹의 복제본으로 만들고, 새 장애 조치 그룹의 오브젝트를 새로 고치고, 원본 계정 역할을 하도록 대상 계정을 승격합니다.

   CREATE FAILOVER GROUP my_stage_failover_group
     AS REPLICA OF myorg.my_account_1.my_stage_failover_group;
   
   ALTER FAILOVER GROUP my_stage_failover_group REFRESH;
   
   ALTER FAILOVER GROUP my_stage_failover_group PRIMARY;
   

   Copy

2. 다음으로, 복제된 스테이지에서 디렉터리 테이블을 새로 고치고 my_stage 의 디렉터리 테이블에서 추적하는 모든 파일을 my_table 이라는 테이블에 복사합니다.

   참고

   COPY INTO 문은 file1 및 file2 를 테이블에 로드하지만 file3 은 로드하지 않습니다. 이는 원본 계정에 file3 을 추가한 후 디렉터리 테이블이 새로 고쳐지지 않았기 때문입니다.

   ALTER STAGE my_stage REFRESH;
   
   COPY INTO my_table FROM @my_stage;
   

   Copy

예 2: 외부 스테이지 및 저장소 통합 복제¶

이 예에서는 외부 스테이지 및 저장소 통합을 대상 계정에 복제하는 샘플 워크플로를 제공합니다.

이 예에서는 Amazon S3 버킷에 대한 보안 액세스 구성하기 를 이미 완료했다고 가정합니다.

이 예의 첫 번째 부분에서는 원본 계정에서 다음 작업을 완료합니다.

1. 데이터베이스 my_database_2 의 Amazon S3 버킷에 대한 저장소 통합을 생성합니다.

   CREATE STORAGE INTEGRATION my_storage_int
     TYPE = external_stage
     STORAGE_PROVIDER = 's3'
     STORAGE_ALLOWED_LOCATIONS = ('s3://mybucket/path')
     STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/blockedpath')
     ENABLED = true;
   

   Copy

2. 저장소 통합 my_storage_int 를 사용하여 데이터베이스 my_database_2 에 외부 스테이지를 생성합니다.

   CREATE STAGE my_ext_stage
     URL = 's3://mybucket/path'
     STORAGE_INTEGRATION = my_storage_int
   

   Copy

3. 장애 조치 그룹을 생성하고 데이터베이스 my_database_2 및 저장소 통합 오브젝트를 포함합니다.

   CREATE FAILOVER GROUP my_external_stage_fg
     OBJECT_TYPES = databases, integrations
     ALLOWED_INTEGRATION_TYPES = storage integrations
     ALLOWED_DATABASES = my_database_2
     ALLOWED_ACCOUNTS = myorg.my_account_2;
   

   Copy

이 예의 두 번째 부분에서는 대상 계정에서 복제 및 장애 조치 프로세스를 완료합니다.

1. 장애 조치 그룹을 원본 계정에 있는 장애 조치 그룹의 복제본으로서 생성하고 새로 고칩니다.

   CREATE FAILOVER GROUP my_external_stage_fg
     AS REPLICA OF myorg.my_account_1.my_external_stage_fg;
   
   ALTER FAILOVER GROUP my_external_stage_fg REFRESH;
   

   Copy

2. 저장소 통합을 대상 계정에 복제한 후에는 클라우드 공급자 권한을 업데이트하여 클라우드 저장소에 대한 복제 통합 액세스 권한을 부여하는 추가 단계를 수행해야 합니다. 자세한 내용은 보조 저장소 통합을 위한 클라우드 저장소 액세스 구성하기 섹션을 참조하십시오.

예 3: 자동 수집 파이프 복제하기¶

이 예에서는 Amazon SQS(Simple Queue Service)와 함께 Amazon SNS(Simple Notification Service) 항목을 사용하여 Snowpipe를 자동화하는 파이프를 복제하는 샘플 워크플로를 제공합니다.

이 예에서는 다음 작업을 이미 완료했다고 가정합니다.

• Amazon S3용 저장소 통합을 생성하고 구성했습니다. 예시 목적으로 my_s3_storage_int 라는 저장소 통합을 사용합니다.

• Amazon SNS 항목 및 구독을 생성하고 SNS 항목에 대해 Snowflake SQS 큐를 구독했습니다.

• 저장소 통합을 참조하는 외부 스테이지를 생성했습니다. 예시 목적으로 my_s3_stage 라는 스테이지를 사용합니다. 자세한 지침은 CREATE STAGE 섹션을 참조하십시오.

원본 계정에서 다음 작업부터 시작합니다.

1. CREATE PIPE 명령을 사용하여 외부 스테이지의 데이터를 mytable 이라는 테이블에 로드하는 자동 수집이 활성화된 파이프를 만듭니다.

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

   Copy

2. 지난 7일 동안의 데이터를 스테이지에서 로드하도록 ALTER PIPE 문으로 파이프를 새로 고칩니다.

   ALTER PIPE mypipe REFRESH;
   

   Copy

3. 마지막으로, CREATE FAILOVER GROUP 을 사용하여 저장소 통합 복제를 허용하는 장애 조치 그룹을 생성합니다.

   CREATE FAILOVER GROUP my_pipe_failover_group
     OBJECT_TYPES = DATABASES, INTEGRATIONS
     ALLOWED_INTEGRATION_TYPES = STORAGE INTEGRATIONS
     ALLOWED_DATABASES = snowpipe_db
     ALLOWED_ACCOUNTS = myorg.my_account_2;
   

   Copy

이 예의 두 번째 부분에서는 대상 계정에서 복제 및 장애 조치 프로세스를 완료합니다.

1. 장애 조치 그룹을 원본 계정에 있는 장애 조치 그룹의 복제본으로서 생성합니다.

   CREATE FAILOVER GROUP my_pipe_failover_group
     AS REPLICA OF myorg.my_account_1.my_pipe_failover_group;
   

   Copy

2. DESCRIBE INTEGRATION 문을 실행하여 보조 배포에서 Snowflake 계정의 AWS IAM 사용자에 대한 ARN을 검색합니다.

   ARN을 사용하여 IAM 사용자에게 S3 버킷에 액세스하는 권한을 부여합니다. 5단계: IAM 사용자에게 버킷 오브젝트에 액세스하는 권한 부여를 참조하십시오.

   DESC INTEGRATION my_s3_storage_int;
   

   Copy

3. SYSTEM$GET_AWS_SNS_IAM_POLICY 시스템 함수를 호출하여 SNS 항목을 구독할 수 있는 새로운 SQS 큐 권한을 부여하는 IAM 정책을 생성합니다. Snowflake는 원본 계정에서 장애 조치 그룹을 복제할 때 대상 계정에 새 SQS 큐를 생성했습니다.

   SELECT SYSTEM$GET_AWS_SNS_IAM_POLICY('<topic_arn>');
   

   Copy

   topic_arn 은 원본 계정의 원본 파이프에 대해 생성한 SNS 항목의 ARN(Amazon Resource Name)입니다.

   그런 다음 SNS 항목에 대해 새 Amazon SQS 큐를 구독합니다.

4. 새 장애 조치 그룹의 오브젝트를 새로 고칩니다.

   ALTER FAILOVER GROUP my_pipe_failover_group REFRESH;
   

   Copy

5. 마지막으로, ALTER FAILOVER GROUP 명령을 사용하여 원본 계정 역할을 하도록 대상 계정을 승격합니다.

   ALTER FAILOVER GROUP my_pipe_failover_group PRIMARY;
   

   Copy

   mypipe 파이프는 원본 계정에서 장애 조치 그룹을 마지막으로 새로 고친 이후 사용 가능한 모든 데이터를 로드하기 시작합니다.

   복제된 파이프가 작동하는지 확인하려면 파이프의 COPY 문에서 테이블을 쿼리하십시오.

   SELECT * FROM mytable;
   

   Copy