Conector do Apache Kafka do Amazon Athena

O conector do Amazon Athena para o Apache Kafka possibilita que o Amazon Athena execute consultas SQL em seus tópicos do Apache Kafka. Use esse conector para visualizar os tópicos e as mensagens do Apache Kafka no Athena como tabelas e linhas, respectivamente.

Esse conector não usa o Glue Connections para centralizar as propriedades de configuração no Glue. A configuração da conexão é feita por meio do Lambda.

Pré-requisitos

Implante o conector na sua Conta da AWS usando o console do Athena ou o AWS Serverless Application Repository. Para ter mais informações, consulte Criar uma conexão de fonte de dados ou Usar o AWS Serverless Application Repository para implantar um conector de fonte de dados.

Limitações

Termos

Compatibilidade de cluster

O conector do Kafka pode ser usado com os seguintes tipos de cluster:

Conectando-se ao Confluent

A conexão com o Confluent requer as etapas a seguir:

Métodos de autenticação compatíveis

O conector é compatível com os métodos de autenticação a seguir.

Para ter mais informações, consulte Configurar a autenticação do conector do Kafka para o Athena.

Formatos de dados de entrada compatíveis

O conector é compatível com os seguintes formatos de dados de entrada:

Parâmetros

Use os parâmetros nesta seção para configurar o conector do Kafka para o Athena.

Suporte ao tipo de dados

A tabela a seguir apresenta os tipos de dados correspondentes compatíveis com o Kafka e o Apache Arrow.

Partições e divisões

Os tópicos do Kafka são divididos em partições. Cada partição é ordenada. Cada mensagem em uma partição tem um ID incremental denominado deslocamento. Cada partição do Kafka é separada em diversas divisões para o processamento paralelo. Os dados estão disponíveis para o período de retenção configurado nos clusters do Kafka.

Práticas recomendadas

Como prática recomendada, use o empilhamento de predicados ao consultar o Athena, como mostrado nos exemplos a seguir.

SELECT * 
FROM "kafka_catalog_name"."glue_schema_registry_name"."glue_schema_name" 
WHERE integercol = 2147483647

SELECT * 
FROM "kafka_catalog_name"."glue_schema_registry_name"."glue_schema_name" 
WHERE timestampcol >= TIMESTAMP '2018-03-25 07:30:58.878'

Como configurar o conector do Kafka

Antes de usar o conector, você deve configurar seu cluster do Apache Kafka. Use o registro de esquemas do AWS Glue para definir seu esquema e configurar a autenticação para o conector.

Ao trabalhar com o registro de esquemas do AWS Glue, preste atenção aos seguintes pontos:

{
  "topicName": "datatypejson",
  "message": {
    "dataFormat": "json",
    "fields": [
      {
        "name": "intcol",
        "mapping": "intcol",
        "type": "INTEGER"
      },
      {
        "name": "varcharcol",
        "mapping": "varcharcol",
        "type": "VARCHAR"
      },
      {
        "name": "booleancol",
        "mapping": "booleancol",
        "type": "BOOLEAN"
      },
      {
        "name": "bigintcol",
        "mapping": "bigintcol",
        "type": "BIGINT"
      },
      {
        "name": "doublecol",
        "mapping": "doublecol",
        "type": "DOUBLE"
      },
      {
        "name": "smallintcol",
        "mapping": "smallintcol",
        "type": "SMALLINT"
      },
      {
        "name": "tinyintcol",
        "mapping": "tinyintcol",
        "type": "TINYINT"
      },
      {
        "name": "datecol",
        "mapping": "datecol",
        "type": "DATE",
        "formatHint": "yyyy-MM-dd"
      },
      {
        "name": "timestampcol",
        "mapping": "timestampcol",
        "type": "TIMESTAMP",
        "formatHint": "yyyy-MM-dd HH:mm:ss.SSS"
      }
    ]
  }
}

{
  "topicName": "datatypecsvbulk",
  "message": {
    "dataFormat": "csv",
    "fields": [
      {
        "name": "intcol",
        "type": "INTEGER",
        "mapping": "0"
      },
      {
        "name": "varcharcol",
        "type": "VARCHAR",
        "mapping": "1"
      },
      {
        "name": "booleancol",
        "type": "BOOLEAN",
        "mapping": "2"
      },
      {
        "name": "bigintcol",
        "type": "BIGINT",
        "mapping": "3"
      },
      {
        "name": "doublecol",
        "type": "DOUBLE",
        "mapping": "4"
      },
      {
        "name": "smallintcol",
        "type": "SMALLINT",
        "mapping": "5"
      },
      {
        "name": "tinyintcol",
        "type": "TINYINT",
        "mapping": "6"
      },
      {
        "name": "floatcol",
        "type": "DOUBLE",
        "mapping": "7"
      }
    ]
  }
}

{
    "type": "record",
    "name": "avrotest",
    "namespace": "example.com",
    "fields": [{
            "name": "id",
            "type": "int"
        },
        {
            "name": "name",
            "type": "string"
        }
    ]
}

syntax = "proto3";
message protobuftest {
string name = 1;
int64 calories = 2;
string colour = 3;
}

{
  "topicName": "datatypejson",
  "message": {
    "dataFormat": "json",
    "fields": [
      {
        "name": "intcol",
        "mapping": "intcol",
        "type": "INTEGER"
      },
      {
        "name": "varcharcol",
        "mapping": "varcharcol",
        "type": "VARCHAR"
      },
      {
        "name": "booleancol",
        "mapping": "booleancol",
        "type": "BOOLEAN"
      },
      {
        "name": "bigintcol",
        "mapping": "bigintcol",
        "type": "BIGINT"
      },
      {
        "name": "doublecol",
        "mapping": "doublecol",
        "type": "DOUBLE"
      },
      {
        "name": "smallintcol",
        "mapping": "smallintcol",
        "type": "SMALLINT"
      },
      {
        "name": "tinyintcol",
        "mapping": "tinyintcol",
        "type": "TINYINT"
      },
      {
        "name": "datecol",
        "mapping": "datecol",
        "type": "DATE",
        "formatHint": "yyyy-MM-dd"
      },
      {
        "name": "timestampcol",
        "mapping": "timestampcol",
        "type": "TIMESTAMP",
        "formatHint": "yyyy-MM-dd HH:mm:ss.SSS"
      }
    ]
  }
}

{
  "topicName": "datatypecsvbulk",
  "message": {
    "dataFormat": "csv",
    "fields": [
      {
        "name": "intcol",
        "type": "INTEGER",
        "mapping": "0"
      },
      {
        "name": "varcharcol",
        "type": "VARCHAR",
        "mapping": "1"
      },
      {
        "name": "booleancol",
        "type": "BOOLEAN",
        "mapping": "2"
      },
      {
        "name": "bigintcol",
        "type": "BIGINT",
        "mapping": "3"
      },
      {
        "name": "doublecol",
        "type": "DOUBLE",
        "mapping": "4"
      },
      {
        "name": "smallintcol",
        "type": "SMALLINT",
        "mapping": "5"
      },
      {
        "name": "tinyintcol",
        "type": "TINYINT",
        "mapping": "6"
      },
      {
        "name": "floatcol",
        "type": "DOUBLE",
        "mapping": "7"
      }
    ]
  }
}

{
    "type": "record",
    "name": "avrotest",
    "namespace": "example.com",
    "fields": [{
            "name": "id",
            "type": "int"
        },
        {
            "name": "name",
            "type": "string"
        }
    ]
}

syntax = "proto3";
message protobuftest {
string name = 1;
int64 calories = 2;
string colour = 3;
}

Configurar a autenticação do conector do Kafka para o Athena

É possível usar uma variedade de métodos para autenticar o cluster do Apache Kafka, como SSL, SASL/SCRAM, SASL/PLAIN e SASL/PLAINTEXT.

A tabela a seguir mostra os tipos de autenticação para o conector, e o protocolo de segurança e o mecanismo SASL para cada um. Para obter mais informações, consulte a seção Segurança da documentação do Apache Kafka.

SSL

Se o cluster for autenticado por SSL, você deverá gerar os arquivos de armazenamento confiável e de armazenamento de chaves, e fazer o upload deles no bucket do Amazon S3. Você deverá fornecer essa referência do Amazon S3 ao implantar o conector. O armazenamento de chaves, o armazenamento confiável e a chave SSL serão armazenados no AWS Secrets Manager. Você fornecerá a chave secreta da AWS ao implantar o conector.

Para obter informações sobre como criar um segredo no Secrets Manager, consulte Criação de um segredo do AWS Secrets Manager.

Para usar esse tipo de autenticação, defina as variáveis ​​de ambiente conforme mostrado na tabela a seguir.

Após a criação de um segredo no Secrets Manager, é possível visualizá-lo no console do Secrets Manager.

Para obter mais informações sobre como usar SSL com o Kafka, consulte Criptografia e autenticação usando SSL na documentação do Apache Kafka.

SASL/SCRAM

Se o cluster usar a autenticação SCRAM, forneça a chave do Secrets Manager associada ao cluster ao implantar o conector. As credenciais da AWS do usuário (chave secreta e chave de acesso) serão usadas para realizar a autenticação com o cluster.

Defina as variáveis ​​de ambiente conforme mostrado na tabela a seguir.

A imagem a seguir apresenta um exemplo de segredo no console do Secrets Manager com dois pares chave/valor: um para username e outro para password.

Para obter mais informações sobre como usar o SASL/SCRAM com o Kafka, consulte Autenticação usando SASL/SCRAM na documentação do Apache Kafka.

Informações de licença

Ao usar esse conector, você reconhece a inclusão de componentes de terceiros, cuja lista pode ser encontrada no arquivo pom.xml desse conector, e concorda com os termos das respectivas licenças de terceiros fornecidas no arquivo LICENSE.txt em GitHub.com.

Recursos adicionais

Para obter mais informações sobre esse conector, visite o site correspondente em GitHub.com.