Set up Openflow Connector for Kinesis for JSON data format¶

Pré-requisitos¶

1. Certifique-se de ter revisado Sobre a Openflow Connector for Kinesis.

2. Ensure that you have Configuração do Openflow - BYOC or Set up Openflow - Snowflake Deployments.

3. Se você usa Openflow - Snowflake Deployments, garanta que já tenha revisado a configuração dos domínios necessários e concedido acesso a esses domínios para o conector Kinesis.

Configure um fluxo Kinesis¶

Como administrador do AWS, execute as seguintes ações em sua conta 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.

Configure a conta Snowflake¶

Como administrador de conta Snowflake, execute as seguintes tarefas:

1. Crie uma nova função ou use uma função existente e conceda a Privilégios de banco de dados.

2. Crie um banco de dados de destino e um esquema de destino que serão usados para criar tabelas de destino para armazenar os dados.

   1. Se você planeja usar o recurso do conector para criar automaticamente a tabela de destino se ela ainda não existir, certifique-se de que o usuário tenha os privilégios necessários para criar e gerenciar objetos Snowflake:

      Objeto	Privilégio	Notas
      Banco de dados	USAGE
      Esquema	USAGE . CREATE TABLE .	Após a criação dos objetos de nível de esquema, os privilégios CREATE object podem ser revogados.
      Tabela	OWNERSHIP	Necessário somente ao usar o conector Kinesis para ingerir dados em uma tabela existente. . Se o conector criar uma nova tabela de destino para registros do fluxo Kinesis, a função padrão do usuário especificado na configuração se tornará o proprietário da tabela.

      Você pode usar o script a seguir para criar e configurar uma função personalizada (requer SECURITYADMIN ou equivalente):

      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. Crie um novo usuário de serviço Snowflake com o tipo SERVICE.

4. Conceda ao usuário do serviço Snowflake a função que você criou nas etapas anteriores.

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

   Copy

5. Configure com autenticação por pares de chaves para o usuário Snowflake SERVICE a partir da etapa 3.

6. A Snowflake recomenda enfaticamente essa etapa. Configure um gerenciador de segredos compatível com o Openflow, por exemplo, AWS, Azure e Hashicorp, e armazene as chaves públicas e privadas no armazenamento de segredos.

   Nota

   Se, por algum motivo, você não quiser usar um gerenciador de segredos, será responsável por proteger os arquivos de chave pública e chave privada usados para autenticação de pares de chaves de acordo com as políticas de segurança de sua organização.

   1. Depois que o gerenciador de segredos estiver configurado, determine como você se autenticará nele. No AWS, é recomendável que você use a função de instância EC2 associada ao Openflow, pois dessa forma nenhum outro segredo precisa ser mantido.

   2. No Openflow, configurar um provedor de parâmetros associado a este Secrets Manager, a partir do menu de configuração no canto superior direito. Navegue até Controller Settings » Parameter Provider e depois buscar seus valores de parâmetro.

   3. Nesse momento, todas as credenciais podem ser referenciadas com os caminhos de parâmetros associados e nenhum valor sensível precisa ser mantido no 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.

Configuração do conector¶

Como engenheiro de dados, execute as seguintes tarefas para instalar e configurar o conector:

Parameters¶

Esta seção descreve todos os parâmetros do Openflow Connector for Kinesis para formato de dados JSON.

O conector consiste em vários módulos. Para ver o conjunto, clique duas vezes no grupo de processos do conector. Você pode definir os parâmetros para cada módulo no contexto de parâmetro do módulo.

Execute o fluxo¶

1. Clique com o botão direito do mouse no plano e selecione Enable all Controller Services.

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

O conector inicia a ingestão de dados.

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';