Conector MSK do Amazon Athena

O conector do Amazon Athena para o Amazon MSK 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. Para obter informações adicionais, consulte Analyze real-time streaming data in Amazon MSK with Amazon Athena (Analisar dados de streaming em tempo real no Amazon MSK com o Amazon Athena) no blog de big data da AWS.

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 obter 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 MSK pode ser usado com os seguintes tipos de cluster:

Métodos de autenticação compatíveis

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

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 MSK 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 "msk_catalog_name"."glue_schema_registry_name"."glue_schema_name" 
WHERE integercol = 2147483647

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

Configuração do conector MSK

Antes de usar o conector, você deve configurar seu cluster do Amazon MSK. 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"
      }
    ]
  }
}

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

Configuração da autenticação para o conector MSK do Athena

É possível usar uma variedade de métodos para autenticar seu cluster do Amazon MSK, incluindo IAM, SSL, SCRAM e Kafka dedicado.

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 Authentication and authorization for Apache Kafka APIs (Autenticação e autorização para APIs do Apache Kafka) no Guia do desenvolvedor do Amazon Managed Streaming for Apache Kafka.

SASL/IAM

Se o cluster usar a autenticação do IAM, você deverá configurar a política do IAM para o usuário ao configurar o cluster. Para obter mais informações, consulte IAM access control (Controle de acesso do IAM) no Guia do desenvolvedor do Amazon Managed Streaming for Apache Kafka.

Para usar esse tipo de autenticação, defina a variável de ambiente do Lambda auth_type como SASL_SSL_AWS_MSK_IAM para o conector.

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.

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.

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.