使用 Apache Kafka 和 Azure Databricks 進行串流處理

本文說明如何在 Azure Databricks 上執行結構化串流工作負載時，使用 Apache Kafka 作為來源或接收器。

如需更多 Kafka，請參閱 Kafka 文件。

從 Kafka 讀取資料

以下是從 Kafka 串流讀取的範例：

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

Azure Databricks 也支援 Kafka 資料來源的批次讀取語意，如下列範例所示：

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

對於累加式批次載入，Databricks 建議將 Kafka 與 Trigger.AvailableNow 搭配使用。 請參閱設定累加批次處理。

在 Databricks Runtime 13.3 LTS 和更新版本中，Azure Databricks 提供 SQL 函式來讀取 Kafka 資料。 只有 Delta Live Tables 或 Databricks SQL 中的串流 tables 才支援使用 SQL 串流處理。 請參閱 read_kafka table-valued 函式。

設定 Kafka 結構化串流讀取器

Azure Databricks 提供 kafka 關鍵詞作為數據格式，以將 connections 設定為 Kafka 0.10+ 。

以下是 Kafka 的最常見組態：

有多種方式指定要訂閱的主題。 您應該只提供下列其中一個 parameters：

其他值得注意的組態：

如需其他選用組態，請參閱結構化串流 Kafka 整合指南。

Schema 的 Kafka 記錄

Kafka 記錄編號 schema 為：

一律使用 key 將 value 和 ByteArrayDeserializer 還原序列化為位元組陣列。 使用 DataFrame 作業（例如 cast("string")）明確地將索引鍵和 values反序列化。

將資料寫入 Kafka

以下是串流寫入 Kafka 的範例：

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

Azure Databricks 也支援 Kafka 資料接收器的批次寫入語意，如下列範例所示：

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

設定 Kafka 結構化串流寫入器

提供給 DataStreamWriter 的 schema 會與 Kafka 匯入端互動。 您可以使用下列欄位：

以下是在寫入 Kafka 時常見的選項：set

如需其他選用組態，請參閱結構化串流 Kafka 整合指南。

擷取 Kafka 計量

您可以 get 串流查詢在具有 avgOffsetsBehindLatest、maxOffsetsBehindLatest和 minOffsetsBehindLatest 計量之所有已訂閱主題中最新的可用可用 offset 數目的平均值、最小值和最大值。 請參閱以互動方式讀取計量。

Get 查詢進程尚未從訂閱的主題取用的估計位元組總數，方法是檢查 estimatedTotalBytesBehindLatest的值。 此估計值基於在過去 300 秒內處理的批次。 您可以透過將選項 bytesEstimateWindowLength 設定為其他值，來變更估計值所基於的時間範圍。 例如，若要將它 set 為 10 分鐘：

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

如果在筆記本中執行串流，您可以在串流查詢進度儀表板中的 [原始資料] 索引標籤下查看這些計量：

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

使用 SSL 將 Azure Databricks 連線至 Kafka

若要啟用用於 Kafka 的 SSL connections，請遵循 Confluent 文件中的指示，SSL 加密與驗證。 您可以提供此處所述的組態 (以 kafka. 為首碼) 作為選項。 例如，您可以在屬性 kafka.ssl.truststore.location 中指定信任存放區位置。

Databricks 建議您：

• 將您的憑證儲存在雲端物件儲存體中。 您可以將對憑證的存取權限制為僅可以存取 Kafka 的叢集。 請參閱 使用 Unity 進行的數據控管 Catalog。
• 將憑證密碼儲存為祕密範圍中的祕密。

下列範例使用物件儲存位置和 Databricks 祕密來啟用 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>))
)

將 HDInsight 上的 Kafka 連線至 Azure Databricks

1. 建立 HDInsight Kafka 叢集。

   如需相關指示，請參閱透過 Azure 虛擬網路連線至 HDInsight 上的 Kafka。

2. 設定 Kafka 訊息代理程式來公告正確的位址。

   請遵循設定 Kafka 進行 IP 公告中的指示。 如果您在 Azure 虛擬機上自行管理 Kafka，請確定代理程式 advertised.listeners 組態 set 主機的內部 IP。

3. 建立 Azure Databricks 叢集。

4. 將 Kafka 叢集對等互連至 Azure Databricks 叢集。

   請遵循對等互連虛擬網路中的指示。

使用 Microsoft Entra ID 和 Azure 事件中樞進行服務主體驗證

Azure Databricks 支援使用事件中樞服務對 Spark 作業進行驗證。 此驗證是透過具有 Microsoft Entra ID 的 OAuth 來完成。

Azure Databricks 支援在下列計算環境中使用用戶端識別碼和祕密進行 Microsoft Entra ID 驗證：

• 設定了單一使用者存取模式的計算上的 Databricks Runtime 12.2 LTS 和更新版本。
• 設定了共用存取模式的計算上的 Databricks Runtime 14.3 LTS 和更新版本。
• 未使用 Unity Catalog設定的 Delta Live Tables 管線。

Azure Databricks 不支援在任何計算環境或使用 Unity Catalog設定的 Delta Live Tables 管線中，使用憑證進行 Microsoft Entra ID 驗證。

此驗證不適用於共享叢集或 Unity Catalog Delta Live Tables。

設定結構化串流 Kafka 連接器

若要使用 Microsoft Entra ID 執行驗證，您需要下列 values：

• 租用戶識別碼。 您可以在 [Microsoft Entra ID] 服務索引標籤中找到此項。

• clientID (也稱為應用程式識別碼)。

• 用戶端密碼。 在您具有此密碼後，應將其作為祕密新增至 Databricks 工作區。 若要新增此祕密，請參閱祕密管理。

• EventHubs 主題。 您可以在特定的 事件中樞命名空間 頁面的 [實體] 區段下的 [事件中樞] 區段中找到 list 個主題。 若要使用多個主題，您可以在事件中樞層級 set IAM 角色。

• EventHubs 伺服器。 您可以在特定 [事件中樞命名空間] 的概觀頁面上找到此項：

  

此外，若要使用 Entra ID，我們需要告知 Kafka 使用 OAuth SASL 機制 (SASL 是一般通訊協定，而 OAuth 是 SASL「機制」的一種類型)：

• kafka.security.protocol 應為 SASL_SSL
• kafka.sasl.mechanism 應為 OAUTHBEARER
• kafka.sasl.login.callback.handler.class 應是 Java 類別的完整名稱，值為 kafkashaded 表示遮蔽 Kafka 類別的登入回撥處理常式。 如需確切的類別，請參閱下列範例。

範例

接下來，讓我們看看執行中的範例：

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)

處理潛在錯誤

• 不支援串流選項。

  如果您嘗試在以 Unity 設定的 Delta Live 管線 Tables 中使用此驗證機制，您可能會收到下列錯誤 Catalog：

  

  若要解決此錯誤，請使用支援的計算組態。 請參閱使用 Microsoft Entra ID 和 Azure 事件中樞進行服務主體驗證。

• 無法建立新的 KafkaAdminClient。

  此為內部錯誤，在下列任何驗證選項不正確時 Kafka 都會擲回此錯誤：

  • 用戶端識別碼 (也稱為應用程式識別碼)
  • 租用戶識別碼
  • EventHubs 伺服器

  若要解決錯誤，請確認這些選項的 values 正確無誤。

  此外，如果您修改範例中預設提供的組態選項 (系統要求您不修改)，例如 kafka.security.protocol，您可能會看到此錯誤。

• 未傳回任何記錄

  如果嘗試顯示或處理 DataFrame 但未取得結果，您將在 UI 中看到下列內容。

  

  此訊息表示驗證成功，但 EventHubs 未傳回任何資料。 一些可能的 (儘管並不詳盡) 原因包括：

  • 指定的 EventHubs 主題錯誤。
  • startingOffsets 的預設 Kafka 組態選項是 latest，而且目前未透過主題收到任何資料。 您可以 setstartingOffsetstoearliest，從 Kafka 的最早位移開始讀取數據。