Apache Kafka ve Azure Databricks ile akış işleme

Bu makalede, Azure Databricks'te Yapılandırılmış Akış iş yüklerini çalıştırırken Apache Kafka'nın kaynak veya havuz olarak nasıl kullanılabileceği açıklanmaktadır.

Daha fazla Kafka için Kafka belgelerine bakın.

Kafka'dan veri okuma

Aşağıda Kafka'dan okunan bir akış örneği verilmiştir:

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

Azure Databricks, aşağıdaki örnekte gösterildiği gibi Kafka veri kaynakları için toplu okuma semantiğini de destekler:

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

Artımlı toplu yükleme için Databricks, Trigger.AvailableNow ile Kafka'nın kullanılmasını önerir. Bkz . Artımlı toplu işlemeyi yapılandırma.

Databricks Runtime 13.3 LTS ve üzerinde Azure Databricks, Kafka verilerini okumak için bir SQL işlevi sağlar. SQL ile akış yalnızca DLT'de veya Databricks SQL'deki akış tablolarında desteklenir. Bakınız read_kafka tablo değerli fonksiyon.

Kafka Yapılandırılmış Akış okuyucusu yapılandırma

Azure Databricks, Kafka 0.10+ ile bağlantıları yapılandırmak için veri biçimi olarak kafka anahtar sözcüğünü sağlar.

Kafka için en yaygın yapılandırmalar şunlardır:

Abone olunacak konuları belirtmenin birden çok yolu vardır. Şu parametrelerden yalnızca birini sağlamanız gerekir:

Diğer önemli yapılandırmalar:

Diğer isteğe bağlı yapılandırmalar için bkz. Yapılandırılmış Akış Kafka Tümleştirme Kılavuzu.

Kafka kayıtları şeması

Kafka kayıtlarının şeması şöyledir:

key ve value, her zaman ByteArrayDeserializer ile bayt dizisi olarak seri durumdan çıkarılır. Anahtarları ve değerleri açıkça seri durumdan çıkarmak için DataFrame işlemlerini (cast("string")gibi) kullanın.

Kafka'ya veri yazma

Aşağıda Kafka'ya akış yazma örneği verilmiştir:

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

Azure Databricks, aşağıdaki örnekte gösterildiği gibi Kafka veri havuzlarına toplu yazma semantiğini de destekler:

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

Kafka Yapılandırılmış Akış yazıcısını yapılandırma

DataStreamWriter'a sağlanan şema Kafka havuzuyla etkileşim kurar. Aşağıdaki alanları kullanabilirsiniz:

Kafka'ya yazarken ayarlanan yaygın seçenekler şunlardır:

Diğer isteğe bağlı yapılandırmalar için Yapılandırılmış Akış Kafka Tümleştirme Kılavuzu'na bakın.

Kafka ölçümlerini alma

avgOffsetsBehindLatest, maxOffsetsBehindLatestve minOffsetsBehindLatest ölçümleriyle abone olunan tüm konular arasında akış sorgusunun en son kullanılabilir uzaklıkların arkasında olduğu uzaklık sayısının ortalamasını, en düşük ve en yüksek değerini alabilirsiniz. Bkz Etkileşimli Ölçümleri Okuma.

estimatedTotalBytesBehindLatestdeğerini inceleyerek abone olunan konu başlıklarından sorgu işleminin kullanmadığı tahmini toplam bayt sayısını alın. Bu tahmin, son 300 saniye içinde işlenen toplu işlemleri temel alır. Tahminin temel aldığı zaman çerçevesi, seçeneği bytesEstimateWindowLength farklı bir değere ayarlanarak değiştirilebilir. Örneğin, 10 dakikaya ayarlamak için:

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

Akışı bir not defterinde çalıştırıyorsanız, akış sorgusu ilerleme durumu panosundaki Ham Veri sekmesinin altında şu ölçümleri görebilirsiniz:

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

Azure Databricks'i Kafka'ya bağlamak için SSL kullanma

Kafka'ya SSL bağlantılarını etkinleştirmek için, SSLile Şifreleme ve Kimlik Doğrulaması Confluent belgelerindeki yönergeleri izleyin. Burada açıklanan yapılandırmaları, önüne kafka. eklemiş olarak seçenekler şeklinde sağlayabilirsiniz. Örneğin, kafka.ssl.truststore.location özelliğinde güven deposu konumunu belirtirsiniz.

Databricks size şu önerileri önerir:

• Sertifikalarınızı bulut nesne depolama alanında depolayın. Sertifikalara erişimi yalnızca Kafka'ya erişebilen kümelerle kısıtlayabilirsiniz. Unity Kataloğu ile veri yönetimiiçin bkz.
• Sertifika parolalarınızı gizli olarak bir gizli kapsamı içinde depolayın.

Aşağıdaki örnek, SSL bağlantısını etkinleştirmek için nesne depolama konumlarını ve Databricks gizli dizilerini kullanır:

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 üzerinde Kafka'yı Azure Databricks'e bağlama

1. HDInsight Kafka kümesi oluşturun.

   Yönergeler için bkz. Azure Sanal Ağ aracılığıyla HDInsight üzerinde Kafka'ya bağlanma.

2. Kafka aracılarını doğru adresi tanıtacak şekilde yapılandırın.

   Kafka'yı IP reklamları için yapılandırma başlığındaki yönergeleri izleyin. Kafka'yı Azure Sanal Makineler'de kendiniz yönetiyorsanız, aracıların advertised.listeners yapılandırmasının konakların iç IP'sine ayarlandığından emin olun.

3. Azure Databricks kümesi oluşturun.

4. Kafka kümesini Azure Databricks kümesiyle eşleyin.

   Eş sanal ağlar başlığındaki yönergeleri izleyin.

Microsoft Entra Kimliği ve Azure Event Hubs ile Hizmet Sorumlusu kimlik doğrulaması

Azure Databricks, Event Hubs hizmetleriyle Spark işlerinin kimlik doğrulamasını destekler. Bu kimlik doğrulaması Microsoft Entra Id ile OAuth aracılığıyla yapılır.

Azure Databricks, aşağıdaki işlem ortamlarında istemci kimliği ve gizli dizi ile Microsoft Entra Id kimlik doğrulamasını destekler:

• Ayrılmış erişim modu (eski adıyla tek kullanıcı erişim modu) ile yapılandırılmış hesaplamada Databricks Runtime 12.2 LTS ve üstü.
• Databricks Runtime 14.3 LTS ve üzeri, standart erişim modu (eski adıyla paylaşılan erişim modu) ile yapılandırılmış hesaplama platformu üzerinde.
• Unity Kataloğu olmadan yapılandırılmış DLT işlem hatları.

Azure Databricks, herhangi bir işlem ortamında veya Unity Kataloğu ile yapılandırılmış DLT işlem hatlarında bir sertifikayla Microsoft Entra Id kimlik doğrulamasını desteklemez.

Bu kimlik doğrulaması, standart erişim moduyla işlemde veya Unity Kataloğu DLT'de çalışmaz.

Yapılandırılmış Akış Kafka Bağlayıcısını Yapılandırma

Microsoft Entra Id ile kimlik doğrulaması gerçekleştirmek için aşağıdaki değerlere ihtiyacınız vardır:

• Kiracı kimliği. Bunu Microsoft Entra Id services sekmesinde bulabilirsiniz.

• Bir clientID (Uygulama Kimliği olarak da bilinir).

• İstemci sırrı. Bunu aldıktan sonra Databricks Çalışma Alanınıza gizli olarak eklemeniz gerekir. Bu sırrı eklemek için bkz Sır yönetimi.

• Bir EventHubs başlığı. Konu listesini, belirli bir Event Hubs Ad Alanı sayfasındaki Varlıklar bölümünün altındaki Event Hubs bölümünde bulabilirsiniz. Birden çok konu başlığıyla çalışmak için IAM rolünü Event Hubs düzeyinde ayarlayabilirsiniz.

• EventHubs sunucusu. Bunu, belirli Event Hubs ad alanınızın genel bakış sayfasında bulabilirsiniz:

  

Ayrıca, Entra Kimliğini kullanmak için Kafka'ya OAuth SASL mekanizmasını kullanmasını söylememiz gerekir (SASL genel bir protokoldür ve OAuth bir SASL "mekanizması" türüdür):

• kafka.security.protocol Olmalıdır SASL_SSL
• kafka.sasl.mechanism Olmalıdır OAUTHBEARER
• kafka.sasl.login.callback.handler.class Java sınıfının eksiksiz tanımlı adı, gölgeli Kafka sınıfımızın oturum açma geri çağırma işleyicisi için kafkashaded değeri olmalıdır. Tam sınıf için aşağıdaki örne bakın.

Örnek

Şimdi çalışan bir örneğe bakalım:

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)

Olası hataları işleme

• Akış seçenekleri desteklenmez.

  Bu kimlik doğrulama mekanizmasını Unity Kataloğu ile yapılandırılmış bir DLT işlem hattında kullanmayı denerseniz aşağıdaki hatayı alabilirsiniz:

  

  Bu hatayı çözmek için desteklenen bir işlem yapılandırması kullanın. Bkz. Microsoft Entra Id ve Azure Event Hubs ile Hizmet Sorumlusu kimlik doğrulaması.

• Yeni bir KafkaAdminClient oluşturulamadı.

  Bu, aşağıdaki kimlik doğrulama seçeneklerinden herhangi biri yanlışsa Kafka'nın attığı bir iç hatadır:

  • İstemci Kimliği (Uygulama Kimliği olarak da bilinir)
  • Kiracı kimliği
  • EventHubs sunucusu

  Hatayı çözmek için bu seçenekler için değerlerin doğru olduğunu doğrulayın.

  Ayrıca, örnekte varsayılan olarak sağlanan yapılandırma seçeneklerini değiştirirseniz (değiştirmemenizi istenirse) bu hatayı görebilirsiniz. Örneğin kafka.security.protocol.

• Döndürülen kayıt yok

  DataFrame'inizi görüntülemeye veya işlemeye çalışıyor ancak sonuç almıyorsanız, kullanıcı arabiriminde aşağıdakileri görürsünüz.

  

  Bu ileti, kimlik doğrulamasının başarılı olduğu ancak EventHubs'ın veri döndürmediğini gösterir. Bazı olası nedenler (hiçbir şekilde kapsamlı olmasa da) şunlardır:

  • Yanlış EventHubs konusunu belirttiniz.
  • Kafka için varsayılan yapılandırma seçeneği startingOffsets şeklindedir: latest, ve şu anda konu başlığı üzerinden henüz herhangi bir veri almıyorsunuz. Kafka’nın ilk ofsetlerinden başlayarak verileri okumaya başlamak için startingOffsetstoearliest’ı ayarlayabilirsiniz.