Configure o Openflow Connector for Kinesis¶

Pré-requisitos¶

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

2. Certifique-se de ter configurado o Openflow.

Configure um fluxo Kinesis¶

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

1. Certifique-se de que você tenha uma conta AWS com permissões de IAM para acessar o Kinesis Streams e DynamoDB.

2. Opcionalmente, crie um fluxo Kinesis de fila de mensagens não entregues (DLQ). As mensagens que não podem ser analisadas com sucesso podem ser redirecionadas para uma DLQ designada.

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_1;
      GRANT USAGE ON DATABASE kinesis_db TO ROLE kinesis_connector_role_1;
      GRANT USAGE ON SCHEMA kinesis_schema TO ROLE kinesis_connector_role_1;
      GRANT CREATE TABLE ON SCHEMA kinesis_schema TO ROLE kinesis_connector_role_1;
      
      -- Only for existing tables
      GRANT OWNERSHIP ON TABLE existing_table1 TO ROLE kinesis_connector_role_1;
      

      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_1 TO USER kinesis_connector_user_1;
   ALTER USER kinesis_connector_user_1 SET DEFAULT_ROLE = kinesis_connector_role_1;
   

   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, configure um provedor de parâmetros associado a esse gerenciador de segredos, no menu de três traços no canto superior direito. Navegue até Controller Settings » Parameter Provider e, em seguida, busque os valores dos parâmetros.

   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. Se qualquer outro usuário Snowflake precisar de acesso aos documentos e tabelas brutos ingeridos pelo conector (por exemplo, para processamento personalizado no Snowflake), conceda a esses usuários a função criada na etapa 1.

8. Designar um warehouse para o conector usar. Comece com o menor tamanho de warehouse e, em seguida, experimente o tamanho, dependendo do número de tabelas sendo replicadas e da quantidade de dados transferidos. Os números de tabelas grandes normalmente são mais bem dimensionados com warehouses multicluster, em vez de warehouses maiores.

Configuração do conector¶

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

Execute o fluxo¶

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

2. Clique com o botão direito do mouse no grupo de processos importado e selecione Start.

O conector inicia a ingestão de dados.

{
    "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"
}

Problemas conhecidos¶

• O grupo de processos do conector tem uma porta de saída denominada “Falha de upload”. Ela pode ser usada para lidar com FlowFiles que não foram carregados com sucesso no Snowflake. Se essa porta não estiver conectada fora do grupo de processos do conector, ela exibirá um sinal de aviso que pode ser ignorado.

• Todos os processadores, quando parados, podem ser ordenados para funcionar uma vez. O processador ConsumeKinesisStream, devido à arquitetura interna, não fará nenhum trabalho significativo quando solicitado a ser executado uma vez. Para que o processador comece a funcionar, ele precisa ser iniciado e funcionar por cerca de dois minutos.