Configuração de Openflow Connector for Amazon Kinesis Data Streams¶

Pré-requisitos¶

1. Consulte Openflow Connector for Amazon Kinesis Data Streams.

2. Certifique-se de ter revisado Configuração do Openflow - BYOC ou Configuração do Openflow – Implantações do Snowflake.

3. Se você usa o Openflow – Implantações do Snowflake, certifique-se de ter revisado a configuração dos domínios necessários e concedido acesso aos domínios necessários para o conector Kinesis.

Configurar funções e políticas do IAM na AWS¶

Como administrador do AWS, execute as seguintes ações em sua conta AWS:

1. Crie um usuário ou função do IAM na AWS que o Openflow usará para acessar o fluxo de dados do Kinesis. Para obter mais informações, consulte como criar usuários do IAM na documentação da AWS.

2. Garanta que o usuário da AWS tenha credenciais de chave de acesso configuradas.

3. Conceda ao usuário da AWS as seguintes permissões do IAM:

   Serviço	Ações	Recursos (ARNs)	Objetivo
   Amazon Kinesis Data Streams	kinesis:DescribeStream, kinesis:DescribeStreamConsumer, kinesis:GetRecords, kinesis:GetShardIterator, kinesis:ListShards, kinesis:RegisterStreamConsumer	arn:aws:kinesis:${REGION}:${ACCOUNT_ID}:stream/${STREAM_NAME}	Descobre fragmentos, lê registros por meio da sondagem de taxa de transferência compartilhada, resolve o ARN do fluxo, registra um consumidor de fan-out aprimorado e sonda o status do consumidor durante o registro.
   Amazon Kinesis Data Streams	kinesis:DeregisterStreamConsumer, kinesis:DescribeStreamConsumer, kinesis:SubscribeToShard	arn:aws:kinesis:${REGION}:${ACCOUNT_ID}:stream/${STREAM_NAME}/consumer/*	Descreve, assina e cancela o registro de consumidores de fan-out aprimorado por ARN de consumidor.
   Amazon DynamoDB	dynamodb:CreateTable, dynamodb:DeleteTable, dynamodb:DescribeTable, dynamodb:GetItem, dynamodb:PutItem, dynamodb:Query, dynamodb:Scan, dynamodb:UpdateItem	arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/${APPLICATION_NAME}, arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/${APPLICATION_NAME}_migration	Cria e gerencia a tabela de ponto de verificação/concessão (concessão de fragmentos, pulsações de nós, pontos de verificação) e uma tabela de migração temporária usada durante a migração única de tabelas de ponto de verificação legadas.

   See moreShow less

   Expand

   Exemplo de política do IAM:

   CopyExpand

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "KinesisStreamAccess",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DescribeStream",
                   "kinesis:DescribeStreamConsumer",
                   "kinesis:GetRecords",
                   "kinesis:GetShardIterator",
                   "kinesis:ListShards",
                   "kinesis:RegisterStreamConsumer"
               ],
               "Resource": "arn:aws:kinesis:${REGION}:${ACCOUNT_ID}:stream/${STREAM_NAME}"
           },
           {
               "Sid": "KinesisConsumerAccess",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DeregisterStreamConsumer",
                   "kinesis:DescribeStreamConsumer",
                   "kinesis:SubscribeToShard"
               ],
               "Resource": "arn:aws:kinesis:${REGION}:${ACCOUNT_ID}:stream/${STREAM_NAME}/consumer/*"
           },
           {
               "Sid": "DynamoDBTableAccess",
               "Effect": "Allow",
               "Action": [
                   "dynamodb:CreateTable",
                   "dynamodb:DeleteTable",
                   "dynamodb:DescribeTable",
                   "dynamodb:GetItem",
                   "dynamodb:PutItem",
                   "dynamodb:Query",
                   "dynamodb:Scan",
                   "dynamodb:UpdateItem"
               ],
               "Resource": [
                   "arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/${APPLICATION_NAME}",
                   "arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/${APPLICATION_NAME}_migration"
               ]
           }
       ]
   }
   

   Show lessSee more

   Scroll to top

   Antes de usar a política de exemplo, substitua os seguintes espaços reservados:

   Espaço reservado	Descrição
   ${REGION}	Sua região AWS (por exemplo, us-east-1)
   ${ACCOUNT_ID}	O ID da sua conta AWS (por exemplo, 123456789012)
   ${STREAM_NAME}	O valor do parâmetro de conector AWS Kinesis Stream Name
   ${APPLICATION_NAME}	O valor do parâmetro de conector AWS Kinesis Application Name. Usado como o nome da tabela de ponto de verificação do DynamoDB e como o nome do consumidor registrado de fan-out aprimorado.

   See moreShow less

   Expand

   Nota

   • ${APPLICATION_NAME}_migration é uma tabela do DynamoDB temporária criada somente durante uma migração única das tabelas de ponto de verificação legadas para o novo esquema. Ela é excluída automaticamente quando a migração é concluída. Se a sua implantação nunca usou o conector baseado em KCL legado, você pode omitir o ARN da tabela de migração da política.

   • A ação dynamodb:DeleteTable é usada durante o processo de migração e pode ser removida da política depois que a migração é confirmada como concluída.

   • A ação kinesis:DeregisterStreamConsumer é invocada quando o processador é removido da tela. Se a entidade do IAM não tiver essa permissão, o registro do consumidor deverá ser cancelado manualmente por meio do console ou da CLI da AWS.

Configure a conta Snowflake¶

Como administrador de conta Snowflake, execute as seguintes tarefas:

1. Crie um novo usuário de serviço Snowflake com o tipo SERVICE.

2. Crie uma nova função ou use uma função existente e conceda os privilégios de banco de dados.

   O conector exige que o usuário crie a tabela de destino. Certifique-se de que o usuário tenha os privilégios necessários para gerenciar objetos Snowflake:

   Objeto	Privilégio	Notas
   Banco de dados	USAGE
   Esquema	USAGE
   Tabela	OWNERSHIP	Necessário para que o conector faça a ingestão de dados em uma tabela.

   See moreShow less

   Expand

   A Snowflake recomenda a criação de um usuário e uma função separados para cada fluxo do Kinesis para ter melhor controle de acesso.

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

   CopyExpand

   USE ROLE securityadmin;
   
   CREATE ROLE openflow_kinesis_connector_role_1;
   GRANT USAGE ON DATABASE kinesis_db TO ROLE openflow_kinesis_connector_role_1;
   GRANT USAGE ON SCHEMA kinesis_schema TO ROLE openflow_kinesis_connector_role_1;
   

   Show lessSee more

   Scroll to top

   Nota

   Os privilégios devem ser concedidos diretamente à função do conector e não podem ser herdados.

3. Configurar a tabela de destino

   Recomendamos fortemente usar a evolução de esquema do lado do servidor para alterações de esquema e uma tabela de erros para registro de erros DML.

   O exemplo abaixo mostra como criar uma tabela e adicionar permissões OWNERSHIP.

   CopyExpand

   USE ROLE openflow_kinesis_connector_role_1;
   
   CREATE TABLE kinesis_db.kinesis_schema.<DESTINATION_TABLE_NAME> (
     kinesisMetadata object
   )
   ENABLE_SCHEMA_EVOLUTION = TRUE
   ERROR_LOGGING = TRUE;
   
   USE ROLE securityadmin;
   GRANT OWNERSHIP ON TABLE <DESTINATION_TABLE_NAME> TO ROLE openflow_kinesis_connector_role_1;
   

   Show lessSee more

   Scroll to top

   Esses conectores são compatíveis com a detecção e a evolução automáticas de esquema. A estrutura das tabelas no Snowflake é definida e evoluída automaticamente para oferecer suporte à estrutura dos novos dados carregados pelo conector. Ela mapeia automaticamente as chaves de primeiro nível do conteúdo do registro para as colunas de tabela correspondentes por nome (sem distinção entre maiúsculas e minúsculas).

   Quando a evolução de esquema está habilitada, o Snowflake pode expandir automaticamente a tabela de destino adicionando novas colunas que são detectadas no fluxo de entrada e descartando as restrições NOT NULL para acomodar novos padrões de dados. Para obter mais informações, consulte Evolução do esquema de tabela.

   Se ENABLE_SCHEMA_EVOLUTION não estiver habilitada, será necessário criar o esquema manualmente estendendo a definição da tabela. O conector tenta corresponder as chaves de primeiro nível do conteúdo do registro às colunas da tabela por nome. Se as chaves do JSON não corresponderem às colunas da tabela, o conector vai ignorá-las.

4. (Opcional) Configurar um gerenciador de segredos

   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.

   1. Depois que o gerenciador de segredos estiver configurado, determine como você se autenticará nele. Na 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. Na tela do Openflow, configure um provedor de parâmetros associado a esse gerenciador de segredos, usando o menu de opções 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.

5. Conceder acesso a usuários

   Outros usuários do Snowflake que precisem de acesso aos dados brutos ingeridos pelo conector (por exemplo, para processamento personalizado no Snowflake) devem receber a função criada na etapa 2.

Configuração do conector¶

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

Concessão de uso em um volume externo¶

Por exemplo, se a tabela Iceberg usar o volume externo kinesis_external_volume e o conector usar a função openflow_kinesis_connector_role_1, execute a seguinte instrução:

USE ROLE ACCOUNTADMIN;
GRANT USAGE ON EXTERNAL VOLUME kinesis_external_volume TO ROLE openflow_kinesis_connector_role_1;

Criar uma tabela Apache Iceberg™ para ingestão¶

O conector não cria tabelas Iceberg automaticamente e não oferece suporte à evolução de esquema. Antes de executar o conector, você deve criar uma tabela Iceberg manualmente.

Ao criar uma tabela Iceberg, é possível usar tipos de dados Iceberg (incluindo VARIANT) ou tipos compatíveis com o Snowflake.

Por exemplo, considere a seguinte mensagem:

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

Para criar uma tabela Iceberg para a mensagem de exemplo, use uma das seguintes instruções:

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,
  name string,
  body_temperature float,
  approved_coffee_types array(string),
  animals_possessed variant,
  date_added date,
  options object(can_walk boolean, can_talk boolean)
)
EXTERNAL_VOLUME = 'my_volume'
CATALOG = 'SNOWFLAKE'
BASE_LOCATION = 'my_location/my_iceberg_table'
ICEBERG_VERSION = 3;