Elaborazione del flusso con Apache Kafka e Azure Databricks

Questo articolo descrive come usare Apache Kafka come origine o sink durante l'esecuzione di carichi di lavoro Structured Streaming in Azure Databricks.

Per altre informazioni su Kafka, vedere la documentazione di Kafka.

Leggere dati da Kafka

Di seguito è riportato un esempio di streaming letto da Kafka:

df = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("subscribe", "<topic>")
  .option("startingOffsets", "latest")
  .load()
)

Azure Databricks supporta anche la semantica di lettura batch per le origini dati Kafka, come illustrato nell'esempio seguente:

df = (spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("subscribe", "<topic>")
  .option("startingOffsets", "earliest")
  .option("endingOffsets", "latest")
  .load()
)

Per il caricamento batch incrementale, Databricks consiglia di usare Kafka con Trigger.AvailableNow. Si veda Configurazione dell'elaborazione batch incrementale.

In Databricks Runtime 13.3 LTS e versioni successive, Azure Databricks fornisce una funzione SQL per la lettura dei dati Kafka. Lo streaming con SQL è supportato solo in tabelle Live Delta o con tabelle di streaming in Databricks SQL. Vedere Funzione con valori di tabella read_kafka.

Configurare il lettore di streaming strutturato Kafka

Azure Databricks fornisce la parola chiave kafka come formato dati per configurare le connessioni a Kafka 0.10+.

Le seguenti sono configurazioni comuni per Kafka:

Esistono diversi modi per specificare gli argomenti da sottoscrivere. È consigliabile specificare solo uno di questi parametri:

Altre configurazioni rilevanti:

Per altre configurazioni facoltative, vedere Guida all'integrazione di Structured Streaming Kafka.

Schema per i record Kafka

Lo schema dei record Kafka è:

key e value vengono sempre deserializzati come matrici di byte con ByteArrayDeserializer. Usare operazioni dataframe (ad esempio cast("string")) per deserializzare in modo esplicito le chiavi e i valori.

Scrivere dati in Kafka

Di seguito è riportato un esempio di scrittura di streaming in Kafka:

(df
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("topic", "<topic>")
  .start()
)

Azure Databricks supporta anche la semantica di scrittura batch nei sink di dati Kafka, come illustrato nell'esempio seguente:

(df
  .write
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("topic", "<topic>")
  .save()
)

Configurare il writer di streaming strutturato Kafka

Lo schema fornito a DataStreamWriter interagisce con il sink Kafka. Puoi usare i seguenti campi:

Di seguito sono riportate le opzioni comuni impostate durante la scrittura in Kafka:

Per altre configurazioni facoltative, vedere Guida all'integrazione di Structured Streaming Kafka.

Recuperare le metriche Kafka

È possibile ottenere la media, min e il numero massimo di offset che la query di streaming è dietro l'offset disponibile più recente tra tutti gli argomenti sottoscritti con le metriche avgOffsetsBehindLatest, maxOffsetsBehindLatest e minOffsetsBehindLatest. Vedere Lettura interattiva delle metriche.

Ottenere il numero totale stimato di byte che il processo di query non ha utilizzato dagli argomenti sottoscritti esaminando il valore di estimatedTotalBytesBehindLatest. Questa stima è basata sui batch elaborati negli ultimi 300 secondi. L'intervallo di tempo su cui si basa la stima può essere modificato impostando l'opzione bytesEstimateWindowLength su un valore diverso. Ad esempio, per impostarlo su 10 minuti:

df = (spark.readStream
  .format("kafka")
  .option("bytesEstimateWindowLength", "10m") # m for minutes, you can also use "600s" for 600 seconds
)

Se si esegue il flusso in un notebook, è possibile visualizzare queste metriche nella scheda Dati non elaborati nel dashboard di stato della query di streaming:

{
  "sources" : [ {
    "description" : "KafkaV2[Subscribe[topic]]",
    "metrics" : {
      "avgOffsetsBehindLatest" : "4.0",
      "maxOffsetsBehindLatest" : "4",
      "minOffsetsBehindLatest" : "4",
      "estimatedTotalBytesBehindLatest" : "80.0"
    },
  } ]
}

Usare SSL per connettere Azure Databricks a Kafka

Per abilitare le connessioni SSL a Kafka, seguire le istruzioni nella documentazione di Confluent Crittografia e autenticazione con SSL. È possibile specificare le configurazioni descritte, precedute da kafka., come opzioni. Ad esempio, si specifica il percorso dell'archivio attendibilità nella proprietà kafka.ssl.truststore.location.

Databricks consiglia:

• Archiviare i certificati nell'archiviazione di oggetti cloud. È possibile limitare l'accesso ai certificati solo ai cluster che possono accedere a Kafka. Vedere Governance dei dati con Catalogo Unity.
• Archiviare le password del certificato come segreti in un ambito segreto.

L'esempio seguente usa i percorsi di archiviazione degli oggetti e i segreti di Databricks per abilitare una connessione SSL:

df = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", ...)
  .option("kafka.security.protocol", "SASL_SSL")
  .option("kafka.ssl.truststore.location", <truststore-location>)
  .option("kafka.ssl.keystore.location", <keystore-location>)
  .option("kafka.ssl.keystore.password", dbutils.secrets.get(scope=<certificate-scope-name>,key=<keystore-password-key-name>))
  .option("kafka.ssl.truststore.password", dbutils.secrets.get(scope=<certificate-scope-name>,key=<truststore-password-key-name>))
)

Connettere Kafka in HDInsight ad Azure Databricks

1. Creare un cluster Kafka in HDInsight.

   Vedere Connettersi a Kafka in HDInsight tramite una rete virtuale di Azure per le istruzioni.

2. Configurare i broker Kafka per annunciare l'indirizzo corretto.

   Seguire le istruzioni in Configurare Kafka per la pubblicità IP. Se si gestisce Kafka in Azure Macchine virtuali, assicurarsi che la configurazione advertised.listeners dei broker sia impostata sull'indirizzo IP interno degli host.

3. Creare un cluster di Azure Databricks.

4. Eseguire il peering del cluster Kafka al cluster Azure Databricks.

   Seguire le istruzioni in Reti virtuali peer.

Autenticazione entità servizio con Microsoft Entra ID e Azure Event Hub

Azure Databricks supporta l'autenticazione dei processi Spark con i servizi di Hub eventi. Questa autenticazione viene eseguita tramite OAuth con Microsoft Entra ID.

Azure Databricks supporta l'autenticazione con ID Microsoft Entra con un ID client e un segreto negli ambienti di calcolo seguenti:

• Databricks Runtime 12.2 LTS e versioni successive su un calcolo configurato con modalità di accesso utente singolo.
• Databricks Runtime 14.3 LTS e versioni successive su un calcolo configurato con modalità di accesso condiviso.
• Pipeline Delta Live Tables configurate senza catalogo Unity.

Azure Databricks non supporta l'autenticazione di Microsoft Entra ID con un certificato in qualsiasi ambiente di calcolo o nelle pipeline delta live tables configurate con catalogo Unity.

Questa autenticazione non funziona nei cluster condivisi o nelle tabelle Live Delta del catalogo Unity.

Configurazione del connettore Kafka Structured Streaming

Per eseguire l'autenticazione con Microsoft Entra ID, sono necessari i valori seguenti:

• Un tenant ID. È possibile trovarla nella scheda Servizi Microsoft Entra ID.

• ID client (noto anche come ID applicazione).

• Segreto client. Una volta ottenuto questo, è necessario aggiungerlo come segreto all'area di lavoro di Databricks. Per aggiungere questo segreto, vedere Gestione dei segreti.

• Argomento di EventHubs. È possibile trovare un elenco di argomenti nella sezione Hub eventi nella sezione Entità in una pagina specifica dello spazio dei nomi di Hub eventi. Per usare più argomenti, è possibile impostare il ruolo IAM a livello di Hub eventi.

• Un server EventHubs. È possibile trovarla nella pagina di panoramica dello spazio dei nomi di Hub eventi specifico:

  

Inoltre, per usare Entra ID, è necessario indicare a Kafka di usare il meccanismo SASL OAuth (SASL è un protocollo generico e OAuth è un tipo di "meccanismo" SASL):

• kafka.security.protocol deve essere SASL_SSL
• kafka.sasl.mechanism deve essere OAUTHBEARER
• kafka.sasl.login.callback.handler.class deve essere un nome completo della classe Java con un valore pari a kafkashaded al gestore di callback di accesso della classe Kafka ombreggiata. Vedere l'esempio seguente per la classe esatta.

Esempio

Si esaminerà quindi un esempio in esecuzione:

Python

# This is the only section you need to modify for auth purposes!
# ------------------------------
tenant_id = "..."
client_id = "..."
client_secret = dbutils.secrets.get("your-scope", "your-secret-name")

event_hubs_server = "..."
event_hubs_topic = "..."
# -------------------------------

sasl_config = f'kafkashaded.org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="{client_id}" clientSecret="{client_secret}" scope="https://{event_hubs_server}/.default" ssl.protocol="SSL";'

kafka_options = {
# Port 9093 is the EventHubs Kafka port
"kafka.bootstrap.servers": f"{event_hubs_server}:9093",
"kafka.sasl.jaas.config": sasl_config,
"kafka.sasl.oauthbearer.token.endpoint.url": f"https://login.microsoft.com/{tenant_id}/oauth2/v2.0/token",
"subscribe": event_hubs_topic,

# You should not need to modify these
"kafka.security.protocol": "SASL_SSL",
"kafka.sasl.mechanism": "OAUTHBEARER",
"kafka.sasl.login.callback.handler.class": "kafkashaded.org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler"
}

df = spark.readStream.format("kafka").options(**kafka_options)

display(df)

Scala

// This is the only section you need to modify for auth purposes!
// -------------------------------
val tenantId = "..."
val clientId = "..."
val clientSecret = dbutils.secrets.get("your-scope", "your-secret-name")

val eventHubsServer = "..."
val eventHubsTopic = "..."
// -------------------------------

val saslConfig = s"""kafkashaded.org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="$clientId" clientSecret="$clientSecret" scope="https://$eventHubsServer/.default" ssl.protocol="SSL";"""

val kafkaOptions = Map(
// Port 9093 is the EventHubs Kafka port
"kafka.bootstrap.servers" -> s"$eventHubsServer:9093",
"kafka.sasl.jaas.config" -> saslConfig,
"kafka.sasl.oauthbearer.token.endpoint.url" -> s"https://login.microsoft.com/$tenantId/oauth2/v2.0/token",
"subscribe" -> eventHubsTopic,

// You should not need to modify these
"kafka.security.protocol" -> "SASL_SSL",
"kafka.sasl.mechanism" -> "OAUTHBEARER",
"kafka.sasl.login.callback.handler.class" -> "kafkashaded.org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler"
)

val scalaDF = spark.readStream
  .format("kafka")
  .options(kafkaOptions)
  .load()

display(scalaDF)

Gestione degli errori potenziali

• Le opzioni di streaming non sono supportate.

  Se si tenta di usare questo meccanismo di autenticazione in una pipeline delta live tables configurata con catalogo Unity, è possibile che venga visualizzato l'errore seguente:

  

  Per risolvere questo errore, usare una configurazione di calcolo supportata. Vedere Autenticazione entità servizio con Microsoft Entra ID e Azure Event Hub.

• Impossibile creare una nuova KafkaAdminClient.

  Si tratta di un errore interno che Kafka genera se una delle opzioni di autenticazione seguenti non è corretta:

  • ID client (detto anche ID applicazione)
  • ID tenant
  • Server EventHubs

  Per risolvere l'errore, verificare che i valori siano corretti per queste opzioni.

  Inoltre, è possibile che venga visualizzato questo errore se si modificano le opzioni di configurazione fornite per impostazione predefinita nell'esempio (a cui è stato chiesto di non modificare), ad esempio kafka.security.protocol.

• Non vengono restituiti record

  Se si sta tentando di visualizzare o elaborare il dataframe ma non si ottengono risultati, nell'interfaccia utente verrà visualizzato quanto segue.

  

  Questo messaggio indica che l'autenticazione ha avuto esito positivo, ma EventHubs non ha restituito dati. Alcuni possibili motivi (anche se non esaustivi) sono:

  • È stato specificato l'argomento EventHubs errato.
  • L'opzione di configurazione Kafka predefinita per startingOffsets è latest e attualmente non si ricevono dati tramite l'argomento. È possibile impostare startingOffsetstoearliest per iniziare a leggere i dati a partire dai primi offset di Kafka.