Set up Openflow Connector for Kinesis for JSON data format¶

전제 조건¶

1. Openflow Connector for Kinesis 정보 을 검토했는지 확인합니다.

2. Ensure that you have Openflow - BYOC 설정 or Set up Openflow - Snowflake Deployments.

3. |OFSFSPCS-plural|를 사용하는 경우 필수 도메인 구성</user-guide/data-integration/openflow/setup-openflow-spcs-sf-allow-list>`을 검토했는지와 :ref:`label-openflow_domains_used_by_openflow_connectors_kinesis 커넥터의 필수 도메인에 대한 액세스 권한을 부여했는지 확인합니다.

Kinesis Stream 설정하기¶

AWS 계정 관리자는 AWS 계정에서 다음 작업을 수행합니다.

1. Ensure that you have an AWS User with IAM permissions to access Kinesis Streams and DynamoDB.

2. Ensure that the AWS User has configured Access Key credentials.

Snowflake 계정 설정하기¶

Snowflake 계정 관리자는 다음 작업을 수행합니다.

1. 새 역할을 생성하거나 기존 역할을 사용하여 데이터베이스 권한 권한을 부여합니다.

2. 데이터를 저장할 대상 데이터베이스와 대상 테이블을 생성하는 데 사용할 대상 스키마를 생성합니다.

   1. 대상 테이블이 없는 경우 커넥터의 기능을 사용하여 대상 테이블을 자동으로 생성할 계획이라면 사용자에게 Snowflake 오브젝트를 생성하고 관리하는 데 필요한 권한이 있는지 확인하십시오.

      오브젝트	권한	참고
      데이터베이스	USAGE
      스키마	USAGE . CREATE TABLE .	스키마 수준 오브젝트가 생성된 후 CREATE object 권한이 취소될 수 있습니다.
      테이블	OWNERSHIP	Kinesis 커넥터를 사용하여 기존 테이블로 데이터를 수집할 때만 필수입니다. . 커넥터가 Kinesis 스트림의 레코드에 대한 새 대상 테이블을 생성하는 경우 구성에 지정된 사용자의 기본 역할이 테이블 소유자가 됩니다.

      다음 스크립트를 사용하여 사용자 지정 역할을 생성하고 구성할 수 있습니다(SECURITYADMIN 또는 이와 동등한 요구 사항 필요):

      USE ROLE SECURITYADMIN;
      
      CREATE ROLE kinesis_connector_role;
      GRANT USAGE ON DATABASE kinesis_db TO ROLE kinesis_connector_role;
      GRANT USAGE ON SCHEMA kinesis_schema TO ROLE kinesis_connector_role;
      GRANT CREATE TABLE ON SCHEMA kinesis_schema TO ROLE kinesis_connector_role;
      
      -- Only for existing tables.
      GRANT OWNERSHIP ON TABLE existing_table TO ROLE kinesis_connector_role;
      

      Copy

3. 유형이 SERVICE 인 새 Snowflake 서비스 사용자를 생성합니다.

4. Snowflake 서비스 사용자에게 이전 단계에서 생성한 역할을 부여합니다.

   GRANT ROLE kinesis_connector_role TO USER kinesis_connector_user;
   ALTER USER kinesis_connector_user SET DEFAULT_ROLE = kinesis_connector_role;
   

   Copy

5. 3단계의 Snowflake SERVICE 사용자에 대해 키 페어 인증 으로 구성합니다.

6. Snowflake는 이 단계를 강력히 권장합니다. Openflow에서 지원하는 시크릿 관리자(예: AWS, Azure, Hashicorp)를 구성하고 공개 및 개인 키를 시크릿 스토어에 저장합니다.

   참고

   어떤 이유로든 시크릿 관리자를 사용하지 않으려면 조직의 보안 정책에 따라 키 페어 인증에 사용되는 공개 키와 개인 키 파일을 보호할 책임이 있습니다.

   1. 시크릿 관리자가 구성되면 인증 방법을 결정합니다. AWS 에서는 다른 시크릿을 유지할 필요가 없으므로 Openflow와 연결된 EC2 인스턴스 역할을 사용하는 것이 좋습니다.

   2. Openflow에서 오른쪽 상단의 햄버거 메뉴에서 이 시크릿 관리자와 연결된 매개 변수 공급자를 구성합니다. Controller Settings » Parameter Provider 로 이동한 다음 매개 변수 값을 가져옵니다.

   3. 이 시점에서 모든 자격 증명은 연결된 매개 변수 경로로 참조할 수 있으며 민감한 값은 Openflow 내에서 유지될 필요가 없습니다.

7. If any other Snowflake users require access to the ingested data and created tables (for example, for custom processing in Snowflake), then grant those users the role created in step 2.

커넥터 설정하기¶

데이터 엔지니어는 다음 작업을 수행하여 커넥터를 설치하고 구성합니다.

Parameters¶

This section describes all parameters for the Openflow Connector for Kinesis for JSON data format.

The connector consists of a several modules. To see the set, double click on the connector process group. You will be able to set the parameters for each module in the module’s parameter context.

플로우 실행¶

1. 평면을 마우스 오른쪽 버튼으로 클릭하고 Enable all Controller Services 를 선택합니다.

2. Right-click on the connector’s process group and select Start.

커넥터가 데이터 수집을 시작합니다.

Iceberg table support¶

Openflow Connector for Kinesis can ingest data into a Snowflake-managed Apache Iceberg™ table when Iceberg Enabled is set to true.

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    kinesisMetadata OBJECT()
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table'
  ENABLE_SCHEMA_EVOLUTION = true;

{
    "id": 1,
    "name": "Steve",
    "body_temperature": 36.6,
    "approved_coffee_types": ["Espresso", "Doppio", "Ristretto", "Lungo"],
    "animals_possessed":
    {
        "dogs": true,
        "cats": false
    },
    "date_added": "2024-10-15"
}

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    kinesisMetadata OBJECT(
        stream STRING,
        shardId STRING,
        approximateArrival STRING,
        partitionKey STRING,
        sequenceNumber STRING,
        subSequenceNumber INTEGER,
        shardedSequenceNumber STRING
    ),
    id INT,
    body_temperature FLOAT,
    name STRING,
    approved_coffee_types ARRAY(STRING),
    animals_possessed OBJECT(dogs BOOLEAN, cats BOOLEAN),
    date_added DATE
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';