Streamverarbeitung mit Apache Kafka und Azure Databricks

In diesem Artikel wird beschrieben, wie Sie Apache Kafka entweder als Quelle oder als Senke verwenden können, wenn Sie Workloads für strukturiertes Streaming in Azure Databricks ausführen.

Mehr über Kafka finden Sie in der Kafka-Dokumentation.

Lesen von Daten aus Kafka

Im Folgenden finden Sie ein Beispiel für einen Streaming-Lesevorgang aus Kafka:

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

Azure Databricks unterstützt auch die Semantik für Batchlesen für Kafka-Datenquellen, wie im folgenden Beispiel gezeigt:

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

Für das inkrementelle Batchladen empfiehlt Databricks die Verwendung von Kafka mit Trigger.AvailableNow. Weitere Informationen finden Sie unter Konfigurieren der inkrementellen Batchverarbeitung.

In Databricks Runtime 13.3 LTS und höher bietet Azure Databricks eine SQL-Funktion zum Lesen von Kafka-Daten. Streaming mit SQL wird nur in Delta Live Tables oder mit Streamingtabellen in Databricks SQL unterstützt. Siehe Tabellenwertfunktion „read_kafka“

Konfigurieren des strukturierter Kafka Streaming-Readers

Azure Databricks stellt das kafka-Schlüsselwort als Datenformat bereit, um Verbindungen mit Kafka 0.10+ zu konfigurieren.

Im Folgenden sind die häufigsten Konfigurationen für Kafka aufgeführt:

Die zu abonnierenden Themen können auf vielfältige Weise angegeben werden. Sie sollten nur einen der folgenden Parameter angeben:

Weitere wichtige Konfigurationen:

Informationen zu weiteren optionalen Konfigurationen finden Sie im Integrationsleitfaden für strukturiertes Streaming und Kafka.

Schema für Kafka-Datensätze

Das Schema der Kafka-Datensätze lautet:

key und value werden mit ByteArrayDeserializer immer als Bytearrays deserialisiert. Verwenden Sie DataFrame-Vorgänge (z. B. cast("string")), um die Schlüssel und Werte explizit zu deserialisieren.

Schreiben von Daten in Kafka

Im Folgenden finden Sie ein Beispiel für einen Streaming-Schreibvorgang auf Kafka:

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

Azure Databricks unterstützt auch die Semantik für Batchlesen für Kafka-Datenquellen, wie im folgenden Beispiel gezeigt:

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

Konfigurieren des strukturierter Kafka Streaming-Writers

Das dem DataStreamWriter bereitgestellte Schema interagiert mit der Kafka-Senke. Sie können die folgenden Felder verwenden:

Im Folgenden finden Sie allgemeine Optionen, die beim Schreiben in Kafka festgelegt werden:

Informationen zu weiteren optionalen Konfigurationen finden Sie im Integrationsleitfaden für strukturiertes Streaming und Kafka.

Abrufen von Kafka-Metriken

Mit den Metriken avgOffsetsBehindLatest, maxOffsetsBehindLatest und minOffsetsBehindLatest können Sie die durchschnittliche, minimale und maximale Anzahl von Offsets, um die die Streamingabfrage hinter dem neuesten verfügbaren Offset zurückliegt, für alle abonnierten Themen abrufen. Weitere Informationen finden Sie unter Interaktives Lesen von Metriken.

Ermitteln Sie die geschätzte Gesamtzahl der Bytes, die der Abfrageprozess noch nicht von den abonnierten Themen verbraucht hat, indem Sie den Wert von estimatedTotalBytesBehindLatest prüfen. Diese Schätzung basiert auf den Batches, die in den letzten 300 Sekunden verarbeitet wurden. Der Zeitrahmen, auf dem die Schätzung basiert, kann geändert werden, indem die Option bytesEstimateWindowLength auf einen anderen Wert gesetzt wird. Beispielsweise kann der Wert auf 10 Minuten gesetzt werden:

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

Wenn Sie den Stream in einem Notebook ausführen, können Sie diese Metriken auf der Registerkarte Rohdaten im Dashboard für den Fortschritt der Streamingabfrage sehen:

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

Verwenden von SSL zum Verbinden von Azure Databricks mit Kafka

Befolgen Sie zum Aktivieren von SSL-Verbindungen mit Kafka die Anweisungen in der Confluent-Dokumentation Verschlüsselung und Authentifizierung mit SSL. Sie können die dort beschriebenen Konfigurationen mit dem Präfix kafka. als Optionen angeben. Beispielsweise geben Sie den Vertrauensspeicherort in der Eigenschaft kafka.ssl.truststore.location an.

Databricks empfiehlt Ihnen Folgendes:

• Speichern Sie Ihre Zertifikate im Cloudobjektspeicher. Sie können den Zugriff auf die Zertifikate nur auf Cluster beschränken, die auf Kafka zugreifen können. Weitere Informationen finden Sie unter Datengovernance mit Unity Catalog.
• Speichern Sie Ihre Zertifikatkennwörter als Geheimnisse in einem Geheimnisbereich.

Im folgenden Beispiel werden Objektspeicherorte und Databricks-Geheimnisse verwendet, um eine SSL-Verbindung zu aktivieren:

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>))
)

Verbinden von Kafka in HDInsight mit Azure Databricks

1. Erstellen Sie einen Kafka-Cluster in HDInsight.

   Anweisungen dazu finden Sie unter Herstellen einer Verbindung mit Kafka in HDInsight über ein virtuelles Azure-Netzwerk.

2. Konfigurieren Sie die Kafka-Broker so, dass sie die richtige Adresse ankündigen.

   Befolgen Sie die Anweisungen unter Konfigurieren von Kafka zum Ankündigen der IP-Adresse. Wenn Sie Kafka selbst in Azure Virtual Machines verwalten, stellen Sie sicher, dass die Konfiguration von advertised.listeners der Broker auf die interne IP-Adresse der Hosts festgelegt ist.

3. Erstellen Sie einen Azure Databricks-Cluster.

4. Nehmen Sie ein Peering des Kafka-Clusters mit dem Azure Databricks-Cluster vor.

   Befolgen Sie die Anweisungen unter Einrichten eines Peerings von virtuellen Netzwerken.

Dienstprinzipal-Authentifizierung mit Microsoft Entra ID und Azure Event Hubs

Azure Databricks unterstützt die Authentifizierung von Spark-Aufträgen mit Event Hubs-Diensten. Die Authentifizierung erfolgt über OAuth mit Microsoft Entra ID.

Azure Databricks unterstützt die Microsoft Entra ID-Authentifizierung mit einer Client-ID und einem Geheimnis in den folgenden Compute-Umgebungen:

• Databricks Runtime 12.2 LTS und höher auf Computeressourcen, die im Zugriffsmodus für Einzelbenutzer*innen konfiguriert sind.
• Databricks Runtime 14.3 LTS und höher auf Computeressourcen, die im Modus für den gemeinsamen Zugriff konfiguriert sind.
• Delta Live Tables-Pipelines, die ohne Unity Catalog konfiguriert sind.

Azure Databricks unterstützt die Microsoft Entra ID-Authentifizierung nicht mit einem Zertifikat in einer Compute-Umgebung oder in Delta Live Tables-Pipelines, die mit Unity Catalog konfiguriert sind.

Diese Authentifizierung funktioniert nicht für freigegebene Cluster oder für Unity Catalog Delta Live Tables.

Konfigurieren des Kafka-Connectors für strukturiertes Streaming

Um die Authentifizierung mit Microsoft Entra ID durchzuführen, benötigen Sie die folgenden Werte:

• Eine Mandanten-ID. Diese finden Sie auf der Registerkarte der Microsoft Entra ID-Dienste.

• Eine Client-ID (auch als Anwendungs-ID bezeichnet).

• Einen geheimen Clientschlüssel. Sobald Sie diesen haben, sollten Sie ihn als Geheimnis zu Ihrem Databricks-Arbeitsbereich hinzufügen. Informationen zum Hinzufügen dieses Geheimnisses finden Sie unter Verwaltung von Geheimnissen.

• Ein EventHubs-Thema. Eine Liste der Themen finden Sie im Abschnitt Event Hubs unter dem Abschnitt Entitäten auf einer bestimmten Seite des Event Hubs-Namespace. Um mit mehreren Themen zu arbeiten, können Sie die IAM-Rolle auf Event Hubs-Ebene festlegen.

• Einen EventHubs-Server. Diesen finden Sie auf der Übersichtsseite Ihres bestimmten Event Hubs-Namespace:

  

Um Entra ID zu verwenden, müssen wir Kafka außerdem anweisen, den OAuth-SASL-Mechanismus zu verwenden (SASL ist ein allgemeines Protokoll, und OAuth ist eine Art von SASL-„Mechanismus“):

• kafka.security.protocol sollte SASL_SSL entsprechen
• kafka.sasl.mechanism sollte OAUTHBEARER entsprechen
• kafka.sasl.login.callback.handler.class sollte ein vollqualifizierter Name der Java-Klasse mit dem Wert kafkashaded für den Anmelderückrufhandler unserer schattierten Kafka-Klasse sein. Das folgende Beispiel zeigt die genaue Klasse.

Beispiel

Als Nächstes sehen wir uns ein laufendes Beispiel an:

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)

Behandeln potenzieller Fehler

• Streaming-Optionen werden nicht unterstützt.

  Wenn Sie versuchen, dieses Authentifizierungsverfahren in einer Delta Live Tables-Pipeline zu verwenden, die mit Unity Catalog konfiguriert ist, wird möglicherweise die folgende Fehlermeldung angezeigt:

  

  Verwenden Sie eine unterstützte Computekonfiguration, um diesen Fehler zu beheben. Weitre Informationen finden Sie unter Dienstprinzipal-Authentifizierung mit Microsoft Entra ID und Azure Event Hubs.

• Fehler beim Erstellen eines neuen KafkaAdminClient.

  Dies ist ein interner Fehler, der von Kafka ausgelöst wird, wenn eine der folgenden Authentifizierungsoptionen falsch ist:

  • Client-ID (auch als Anwendungs-ID bezeichnet)
  • Mandanten-ID
  • EventHubs-Server

  Um den Fehler zu beheben, überprüfen Sie, ob die Werte für diese Optionen korrekt sind.

  Außerdem kann dieser Fehler auftreten, wenn Sie die Konfigurationsoptionen ändern, die im Beispiel standardmäßig vorgesehen sind (und die Sie nicht ändern sollten), wie beispielsweise kafka.security.protocol.

• Es werden keine Datensätze zurückgegeben

  Wenn Sie versuchen, Ihren DataFrame anzuzeigen oder zu verarbeiten, aber keine Ergebnisse erhalten, wird in der Benutzeroberfläche Folgendes angezeigt.

  

  Diese Meldung bedeutet, dass die Authentifizierung erfolgreich war, aber EventHubs keine Daten zurückgegeben hat. Einige mögliche (aber keineswegs vollständige) Gründe sind:

  • Sie haben das falsche EventHubs-Thema angegeben.
  • Die standardmäßige Kafka-Konfigurationsoption für startingOffsets ist latest und Sie erhalten derzeit keine Daten über das Thema. Sie können startingOffsetstoearliest festlegen, damit Sie mit dem Lesen von Daten starten, beginnend mit den frühesten Offsets von Kafka.