Autentizace

Na této stránce jsou uvedeny nejběžnější metody ověřování konektoru Kafka na Azure Databricks.

Úplný seznam podporovaných metod ověřování najdete v dokumentaci kafka. Referenční informace o možnostech ověřování najdete v tématu Ověřování.

Připojte se k Azure Event Hubs pomocí aplikačního nástroje

Azure Databricks podporuje ověřování úloh Sparku se službami Event Hubs pomocí OAuth s Microsoft Entra ID.

Diagram ověřování AAD

Připojení pomocí přihlašovacích údajů služby Katalogu Unity

V Databricks Runtime 16.1 a novějších podporuje Azure Databricks přihlašovací údaje služby Katalogu Unity pro ověřování v Azure Event Hubs. Databricks doporučuje tento přístup, pokud spouštíte streamování Kafka na sdílených clusterech nebo bezserverových výpočetních prostředcích.

Pokud chcete pro ověřování použít přihlašovací údaje služby Katalogu Unity, postupujte takto:

Vytvořte nové přihlašovací údaje služby Katalogu Unity. Viz Vytvoření přihlašovacích údajů služby.
- Ověřte, že přístupový konektor připojený k přihlašovacím údajům služby má správná oprávnění pro připojení k Azure Event Hubs.
Nastavte zdrojovou možnost databricks.serviceCredential na název přihlašovacích údajů služby.

Následující příklad nakonfiguruje Kafka jako zdroj pomocí přihlašovacích údajů služby:

Python

kafka_options = {
  "kafka.bootstrap.servers": "<bootstrap-hostname>:9092",
  "subscribe": "<topic>",
  "databricks.serviceCredential": "<service-credential-name>",
  # Optional: set this only if Databricks can't infer the scope for your Kafka service.
  # "databricks.serviceCredential.scope": "https://<event-hubs-server>/.default",
}

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

Scala

val kafkaOptions = Map(
  "kafka.bootstrap.servers" -> "<bootstrap-hostname>:9092",
  "subscribe" -> "<topic>",
  "databricks.serviceCredential" -> "<service-credential-name>",
  // Optional: set this only if Databricks can't infer the scope for your Kafka service.
  // "databricks.serviceCredential.scope" -> "https://<event-hubs-server>/.default",
)

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

SQL

SELECT * FROM read_kafka(
  bootstrapServers => '<bootstrap-hostname>:9092',
  subscribe => '<topic>',
  serviceCredential => '<service-credential-name>'
);

Poznámka:

Pokud pro připojení k Systému Kafka používáte přihlašovací údaje služby Katalogu Unity, nepoužívejte následující možnosti:

kafka.sasl.mechanism
kafka.sasl.jaas.config
kafka.security.protocol
kafka.sasl.client.callback.handler.class
kafka.sasl.oauthbearer.token.endpoint.url

Připojení s ID klienta a tajným kódem

Azure Databricks podporuje ověřování Microsoft Entra ID s ID klienta a tajným kódem v následujících výpočetních prostředích:

Databricks Runtime 12.2 LTS a vyšší na výpočetních prostředcích nakonfigurovaných pomocí vyhrazeného režimu přístupu
Databricks Runtime 14.3 LTS a vyšší na výpočetních prostředcích nakonfigurovaných pomocí standardního režimu přístupu
Kanály Lakeflow nakonfigurované bez katalogu Unity

Azure Databricks nepodporuje ověřování Microsoft Entra ID pomocí certifikátu v žádném výpočetním prostředí ani v kanálech Lakeflow nakonfigurovaných pomocí katalogu Unity.

Toto ověřování nefunguje na výpočetních prostředcích se standardním režimem přístupu ani v kanálech Unity Catalog Lakeflow.

Pokud chcete provést ověřování pomocí Microsoft Entra ID, musíte mít následující hodnoty:

Identifikátor nájemce. Najdete ho na kartě služby Microsoft Entra ID.
ID klienta, označované také jako ID aplikace.
Tajemství klienta. Přidejte ho jako tajný kód do pracovního prostoru Databricks. Viz správa tajemství.
Téma EventHubs Seznam témat najdete v části Event Hubs pod částí Entity na konkrétní stránce Oboru názvů Event Hubs. Pokud chcete pracovat s několika tématy, můžete nastavit roli IAM na úrovni služby Event Hubs.
Server služby EventHubs. Najdete ho na stránce přehledu vašeho konkrétního oboru názvů služby Event Hubs:

Chcete-li použít Entra ID, musíte v systému Kafka nakonfigurovat použití OAuth SASL:

Nastavte kafka.security.protocol na hodnotu SASL_SSL.
Nastavte kafka.sasl.mechanism na hodnotu OAUTHBEARER.
Nastavte kafka.sasl.login.callback.handler.class na plně kvalifikovaný název třídy Java. Kvalifikovaný název je kafkashaded a obslužná rutina zpětného volání pro přihlášení stínované třídy Kafka od Databricks. Přesné třídy najdete v následujícím příkladu.

SASL je obecný ověřovací protokol a OAuth je mechanismus SASL.

Následující příklad nakonfiguruje kafka pro připojení k Azure Event Hubs pomocí ověřování Microsoft Entra ID s ID klienta a tajným kódem:

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 = {
    "kafka.bootstrap.servers": f"{event_hubs_server}:9093", # Port 9093 is the EventHubs Kafka port
    "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(
  "kafka.bootstrap.servers" -> s"$eventHubsServer:9093", // Port 9093 is the EventHubs Kafka port
  "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)

SQL

CREATE OR REFRESH STREAMING TABLE <table_name>
AS
SELECT * FROM STREAM read_kafka(
  bootstrapServers => '<event-hubs-server>:9093',
  subscribe => '<event-hubs-topic>',
  `kafka.security.protocol` => 'SASL_SSL',
  `kafka.sasl.mechanism` => 'OAUTHBEARER',
  `kafka.sasl.jaas.config` => '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.sasl.oauthbearer.token.endpoint.url` => 'https://login.microsoft.com/<tenant-id>/oauth2/v2.0/token',
  `kafka.sasl.login.callback.handler.class` => 'kafkashaded.org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler'
);

Ověřování pomocí SASL/PLAIN

Pokud se chcete připojit k Kafka pomocí ověřování SASL/PLAIN (uživatelské jméno a heslo), nakonfigurujte následující možnosti. Použijte název stínované PlainLoginModule třídy:

Python

kafka_options = {
  "kafka.bootstrap.servers": "<bootstrap-server>:9093",
  "subscribe": "<topic>",
  "kafka.security.protocol": "SASL_SSL",
  "kafka.sasl.mechanism": "PLAIN",
  "kafka.sasl.jaas.config":
    'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="<username>" password="<password>";',
}

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

Scala

val kafkaOptions = Map(
  "kafka.bootstrap.servers" -> "<bootstrap-server>:9093",
  "subscribe" -> "<topic>",
  "kafka.security.protocol" -> "SASL_SSL",
  "kafka.sasl.mechanism" -> "PLAIN",
  "kafka.sasl.jaas.config" ->
    """kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="<username>" password="<password>";""",
)

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

SQL

SELECT * FROM STREAM read_kafka(
  bootstrapServers => '<bootstrap-server>:9093',
  subscribe => '<topic>',
  `kafka.security.protocol` => 'SASL_SSL',
  `kafka.sasl.mechanism` => 'PLAIN',
  `kafka.sasl.jaas.config` => 'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="<username>" password="<password>";'
);

Azure Databricks doporučuje, abyste heslo uložili jako tajný kód, a ne přímo do kódu. Další informace najdete v Správa tajemství.

Použijte SASL/SCRAM k ověření

Pokud se chcete připojit k Kafka pomocí SASL/SCRAM (SCRAM-SHA-256 nebo SCRAM-SHA-512), nakonfigurujte následující možnosti. Použijte název stínované ScramLoginModule třídy:

Python

kafka_options = {
  "kafka.bootstrap.servers": "<bootstrap-server>:9093",
  "subscribe": "<topic>",
  "kafka.security.protocol": "SASL_SSL",
  "kafka.sasl.mechanism": "SCRAM-SHA-512",
  "kafka.sasl.jaas.config":
    'kafkashaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="<username>" password="<password>";',
}

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

Scala

val kafkaOptions = Map(
  "kafka.bootstrap.servers" -> "<bootstrap-server>:9093",
  "subscribe" -> "<topic>",
  "kafka.security.protocol" -> "SASL_SSL",
  "kafka.sasl.mechanism" -> "SCRAM-SHA-512",
  "kafka.sasl.jaas.config" ->
    """kafkashaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="<username>" password="<password>";""",
)

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

SQL

SELECT * FROM STREAM read_kafka(
  bootstrapServers => '<bootstrap-server>:9093',
  subscribe => '<topic>',
  `kafka.security.protocol` => 'SASL_SSL',
  `kafka.sasl.mechanism` => 'SCRAM-SHA-512',
  `kafka.sasl.jaas.config` => 'kafkashaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="<username>" password="<password>";'
);

Poznámka:

SCRAM-SHA-512 NahraďteSCRAM-SHA-256, pokud je cluster Kafka nakonfigurovaný tak, aby používal SCRAM-SHA-256.

Azure Databricks doporučuje, abyste heslo uložili jako tajný kód, a ne přímo do kódu. Další informace najdete v Správa tajemství.

Použití ssl pro připojení Azure Databricks k Kafka

Pokud chcete povolit připojení SSL/TLS k Kafka, nastavte kafka.security.protocolSSL a poskytněte možnosti konfigurace úložiště důvěryhodnosti a úložiště klíčů s předponou kafka.. Pro připojení SSL, která vyžadují pouze ověřování serveru (jednosměrný protokol TLS), musíte použít úložiště důvěryhodnosti. Pro vzájemné ověřování TLS (mTLS), při kterém broker Kafka také ověřuje klienta, musíte použít jak úložiště důvěryhodných certifikátů, tak úložiště klíčů.

K dispozici jsou následující možnosti protokolu SSL/TLS. Úplný seznam vlastností SSL najdete v dokumentaci ke konfiguraci SSL Apache Kafka a v dokumentaci Confluent v části Šifrování a ověřování pomocí SSL.

Možnost	Description
`kafka.security.protocol`	Nastavte na `SSL` povolení šifrování TLS.
`kafka.ssl.truststore.location`	Cesta k souboru úložiště důvěryhodnosti obsahujícího certifikáty důvěryhodné certifikační autority
`kafka.ssl.truststore.password`	Heslo pro soubor důvěryhodného úložiště.
`kafka.ssl.truststore.type`	Formát souboru důvěryhodného úložiště (výchozí: `JKS`).
`kafka.ssl.keystore.location`	Cesta k souboru úložiště klíčů obsahujícímu klientský certifikát a privátní klíč (vyžaduje se pro mTLS).
`kafka.ssl.keystore.password`	Heslo pro soubor úložiště klíčů.
`kafka.ssl.key.password`	Heslo pro privátní klíč v úložišti klíčů.
`kafka.ssl.endpoint.identification.algorithm`	Ověřovací algoritmus názvu hostitele Výchozí hodnota je `https`. Nastavte na prázdný řetězec, aby se zakázalo.

Pokud používáte PROTOKOL SSL, databricks doporučuje:

Uložte certifikáty do svazku katalogu Unity. Uživatelé, kteří mají přístup ke čtení ze svazku, můžou používat vaše certifikáty Kafka. Další informace najdete v tématu Co jsou svazky katalogu Unity?.
Uložte hesla certifikátů jako tajemství v tajném prostoru. Další informace najdete v tématu Správa tajných oborů.

Následující příklad používá umístění úložiště objektů a tajné kódy Databricks k povolení připojení SSL:

Python

df = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<bootstrap-server>:9093")
  .option("kafka.security.protocol", "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>))
)

Scala

val df = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<bootstrap-server>:9093")
  .option("kafka.security.protocol", "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>))

SQL

SELECT * FROM read_kafka(
  bootstrapServers => '<bootstrap-server>:9093',
  subscribe => '<topic>',
  `kafka.security.protocol` => 'SSL',
  `kafka.ssl.truststore.location` => '<truststore-location>',
  `kafka.ssl.keystore.location` => '<keystore-location>',
  `kafka.ssl.keystore.password` => secret('<certificate-scope-name>', '<keystore-password-key-name>'),
  `kafka.ssl.truststore.password` => secret('<certificate-scope-name>', '<truststore-password-key-name>')
);

Připojení Kafka ve službě HDInsight k Azure Databricks

Vytvořte cluster HDInsight Kafka.

Pokyny najdete v tématu Pojení k Systému Kafka ve službě HDInsight prostřednictvím Azure Virtual Network.
Nakonfigurujte zprostředkovatele Kafka tak, aby inzerovali správnou adresu.

Postupujte podle pokynů v Konfigurujte Kafka pro označení IP. Pokud kafka spravujete sami na Azure Virtual Machines, ujistěte se, že je konfigurace advertised.listeners zprostředkovatelů nastavená na interní IP adresu hostitelů.
Vytvořte cluster Azure Databricks.
Propojit cluster Kafka s clusterm Azure Databricks.

Postupujte podle pokynů v partnerských virtuálních sítích.

Použití stínovaných názvů tříd Kafka v Databricks

Azure Databricks seskupuje proprietární verze klientských knihoven Kafka. Všechny názvy klientských tříd Kafka, na které odkazujete v možnostech konfigurace ověřování, musí místo standardního názvu opensourcové třídy používat předponu názvu stínované třídy. To platí pro všechny třídy odkazované v možnostech, jako kafka.sasl.jaas.config, kafka.sasl.login.callback.handler.class a kafka.sasl.client.callback.handler.class.

Pokud použijete nezastíněné názvy tříd, kód vyvolá RESTRICTED_STREAMING_OPTION_PERMISSION_ENFORCED chybu. Další podrobnosti najdete v nejčastějších dotazech .

Zpracování potenciálních chyb

Vytvoření nového se nezdařilo. KafkaAdminClient

Tato vnitřní chyba Kafka se vyvolá, pokud některá z následujících možností ověřování není správná:
- ID klienta (označované také jako ID aplikace)
- Identifikátor nájemce
- Server Event Hubs
Pokud chcete chybu vyřešit, ověřte správnost hodnot pro tyto možnosti. Kromě toho se tato chyba může zobrazit, pokud upravíte možnosti konfigurace, které jsou ve výchozím nastavení k dispozici v příkladu (například kafka.security.protocol).
Nebyly vráceny žádné záznamy.

Pokud se pokoušíte datový rámec zobrazit nebo zpracovat, ale nezobrazují se vám výsledky, zobrazí se v uživatelském rozhraní následující kód.

Tato zpráva znamená, že ověřování proběhlo úspěšně, ale Služba EventHubs nevrátila žádná data. Některé možné (i když bez vyčerpávajícího) důvodu jsou:
- Zadali jste nesprávné téma EventHubs .
- Výchozí možnost konfigurace Kafka pro startingOffsets je latest a momentálně nedostáváte žádná data prostřednictvím topicu. Můžete nastavit startingOffsets, aby se earliest začalo číst data od nejstarších offsetů Kafky.

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-07-10

Autentizace

Připojte se k Azure Event Hubs pomocí aplikačního nástroje

Připojení pomocí přihlašovacích údajů služby Katalogu Unity

Python

Scala

SQL

Připojení s ID klienta a tajným kódem

Python

Scala

SQL

Ověřování pomocí SASL/PLAIN

Python

Scala

SQL

Použijte SASL/SCRAM k ověření

Python

Scala

SQL

Použití ssl pro připojení Azure Databricks k Kafka

Python

Scala

SQL

Připojení Kafka ve službě HDInsight k Azure Databricks

Použití stínovaných názvů tříd Kafka v Databricks

Zpracování potenciálních chyb

Váš názor

Další materiály