Connettore Apache Kafka di Amazon Athena

Il connettore Amazon Athena per Apache Kafka consente ad Amazon Athena di eseguire query SQL sui tuoi argomenti di Apache Kafka. Utilizza questo connettore per visualizzare gli argomenti di Apache Kafka come tabelle e i messaggi come righe in Athena.

Questo connettore non utilizza Glue Connections per centralizzare le proprietà di configurazione in Glue. La configurazione della connessione viene effettuata tramite Lambda.

Prerequisiti

Implementa il connettore sul tuo Account AWS utilizzando la console Athena o AWS Serverless Application Repository. Per ulteriori informazioni, consulta Creare una connessione a una fonte di dati o Utilizzare il AWS Serverless Application Repository per distribuire un connettore di origine dati.

Limitazioni

Termini

Compatibilità dei cluster

Il connettore Kafka può essere utilizzato con i seguenti tipi di cluster.

Connessione a Confluent

Per la connessione a Confluent sono necessari i seguenti passaggi:

Metodi di autenticazione di supportati

Il connettore supporta i seguenti metodi di autenticazione.

Per ulteriori informazioni, consulta Configurazione dell'autenticazione del connettore Kafka di Athena.

Formati di dati di input supportati

Il connettore supporta i seguenti formati di dati di input.

Parametri

Usa i parametri in questa sezione per configurare il connettore Athena Kafka.

Supporto dei tipi di dati

Nella tabella seguente vengono illustrati i tipi di dati corrispondenti supportati per Kafka e Apache Arrow.

Partizioni e suddivisioni

Gli argomenti di Kafka sono suddivisi in partizioni. Ogni partizione è ordinata. Ogni messaggio in una partizione ha un ID incrementale chiamato offset. Ogni partizione Kafka è ulteriormente suddivisa in più suddivisioni per l'elaborazione in parallelo. I dati sono disponibili per il periodo di conservazione configurato nei cluster Kafka.

Best practice

Come procedura consigliata, utilizza il pushdown del predicato quando esegui query su Athena, come negli esempi seguenti.

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'

Configurazione del connettore Kafka

Prima di poter utilizzare il connettore, è necessario configurare il cluster Apache Kafka, utilizzare il registro degli schemi di AWS Glue per definire lo schema e configurare l'autenticazione del connettore.

Quando utilizzate AWS Glue lo Schema Registry, tenete presente i seguenti punti:

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

Configurazione dell'autenticazione del connettore Kafka di Athena

È possibile utilizzare diversi metodi per l'autenticazione nel cluster Apache Kafka, incluso SSL,. SASL/SCRAM, SASL/PLAIN, and SASL/PLAINTEXT

La tabella seguente mostra i tipi di autenticazione per il connettore, il protocollo di sicurezza e il meccanismo SASL per ciascuno di questi metodi. Per ulteriori informazioni, consulta la sezione Sicurezza nella documentazione di Apache Kafka.

SSL

Se il cluster è autenticato tramite SSL, devi generare i file del trust store e del key store e caricarli nel bucket Amazon S3. È necessario fornire questo riferimento Amazon S3 quando implementi il connettore. Il key store, il trust store e la chiave SSL sono archiviati in AWS Secrets Manager. La chiave AWS segreta viene fornita al momento della distribuzione del connettore.

Per informazioni sulla creazione di un segreto in Secrets Manager, consulta la pagina Crea un segreto AWS Secrets Manager.

Per utilizzare questo tipo di autenticazione, è necessario impostare le variabili di ambiente come illustrato nella tabella seguente.

Dopo avere creato un segreto in Gestione dei segreti, puoi visualizzarlo nella console Gestione dei segreti.

Per ulteriori informazioni sull'utilizzo di SSL con Kafka, consulta Crittografia e autenticazione tramite SSL nella documentazione di Apache Kafka.

SASL/SCRAM

Se il cluster utilizza l'autenticazione SCRAM, fornisci la chiave Gestione dei segreti associata al cluster quando implementi il connettore. Le credenziali AWS dell'utente (chiave segreta e chiave di accesso) vengono utilizzate per autenticarsi al cluster.

Imposta le variabili di ambiente come illustrato nella tabella seguente.

L'immagine seguente mostra un esempio di segreto nella console Gestione dei segreti con due coppie chiave/valore: una per username e una per password.

Per ulteriori informazioni sull'utilizzo di SASL/SCRAM con Kafka, consulta Autenticazione tramite SASL/SCRAM nella documentazione di Apache Kafka.

Informazioni sulla licenza

Utilizzando questo connettore, l'utente riconosce l'inclusione di componenti di terze parti, un elenco dei quali è disponibile nel file pom.xml relativo a questo connettore, e accetta i termini delle rispettive licenze di terze parti fornite nel file LICENSE.txt su GitHub .com.

Risorse aggiuntive

Per ulteriori informazioni su questo connettore, visita il sito corrispondente su GitHub .com.