Streamování z Apache Pulsearu

Důležité

Tato funkce je ve verzi Public Preview.

Ve službě Databricks Runtime 14.1 a novějších můžete pomocí strukturovaného streamování streamovat data z Apache Pulsear v Azure Databricks.

Structured Streaming poskytuje sémantiku zpracování typu exactly-once pro data načtená ze zdrojů Pulsar.

Příklad syntaxe

Následuje základní příklad použití strukturovaného streamování ke čtení z Pulsaru:

Python

query = (spark.readStream
  .format("pulsar")
  .option("service.url", "pulsar://broker.example.com:6650")
  .option("topics", "topic1,topic2")
  .load()
)

Scala

val query = spark.readStream
  .format("pulsar")
  .option("service.url", "pulsar://broker.example.com:6650")
  .option("topics", "topic1,topic2")
  .load()

Chcete-li číst z témat Pulsar, musíte zadat service.url a jednu z následujících možností:

• topic
• topics
• topicsPattern

Úplný seznam možností najdete v tématu Konfigurace možností pro čtení streamu Pulsar.

Přihlaste se k Pulsaru

Azure Databricks podporuje ověřování pomocí truststore a keystore pro Pulsar. Databricks doporučuje používat tajné kódy k ukládání podrobností o konfiguraci.

Úplný seznam možností ověřování najdete v tématu Ověřování.

Example

Následující příklad ukazuje konfiguraci možností ověřování:

Python

client_auth_params = dbutils.secrets.get(scope="pulsar", key="clientAuthParams")
client_pw = dbutils.secrets.get(scope="pulsar", key="clientPw")

# clientAuthParams is a comma-separated list of key-value pairs, such as:
# "keyStoreType:JKS,keyStorePath:/var/private/tls/client.keystore.jks,keyStorePassword:clientpw"

query = (spark.readStream
  .format("pulsar")
  .option("service.url", "pulsar://broker.example.com:6650")
  .option("topics", "topic1,topic2")
  .option("startingOffsets", starting_offsets)
  .option("pulsar.client.authPluginClassName", "org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls")
  .option("pulsar.client.authParams", client_auth_params)
  .option("pulsar.client.useKeyStoreTls", "true")
  .option("pulsar.client.tlsTrustStoreType", "JKS")
  .option("pulsar.client.tlsTrustStorePath", trust_store_path)
  .option("pulsar.client.tlsTrustStorePassword", client_pw)
  .load()
)

Scala

val clientAuthParams = dbutils.secrets.get(scope = "pulsar", key = "clientAuthParams")
val clientPw = dbutils.secrets.get(scope = "pulsar", key = "clientPw")

// clientAuthParams is a comma-separated list of key-value pairs, such as:
// "keyStoreType:JKS,keyStorePath:/var/private/tls/client.keystore.jks,keyStorePassword:clientpw"

val query = spark.readStream
  .format("pulsar")
  .option("service.url", "pulsar://broker.example.com:6650")
  .option("topics", "topic1,topic2")
  .option("startingOffsets", startingOffsets)
  .option("pulsar.client.authPluginClassName", "org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls")
  .option("pulsar.client.authParams", clientAuthParams)
  .option("pulsar.client.useKeyStoreTls", "true")
  .option("pulsar.client.tlsTrustStoreType", "JKS")
  .option("pulsar.client.tlsTrustStorePath", trustStorePath)
  .option("pulsar.client.tlsTrustStorePassword", clientPw)
  .load()

Pulzární schéma

Při čtení z Pulsaru závisí schéma řádků na schématech témat zdroje.

• U témat se schématem Avro nebo JSON se názvy polí a typy polí zachovají ve výsledném datovém rámci Sparku.
• V případě témat bez schématu nebo jednoduchého datového typu v Pulsearu se datová část načte do value sloupce.
• Pokud nakonfigurujete datový tok tak, aby četl více témat s různými schématy, nastavte allowDifferentTopicSchemas tak, aby se nezpracovaný obsah načetl do sloupce value.

Záznamy systému Pulsar mají následující pole metadat:

Sloupec	Typ
__key	binary
__topic	string
__messageId	binary
__publishTime	timestamp
__eventTime	timestamp
__messageProperties	map<String, String>

Nakonfigurujte možnosti pro streamové čtení Pulsar

Úplný seznam možností naleznete v tématu Pulsar.

Vytvoření počátečních posunů JSON

Pokud chcete použít vlastní ID zprávy, které určuje posun, jako JSON s startingOffsets možností, podívejte se na následující příklad:

import org.apache.spark.sql.pulsar.JsonUtils
import org.apache.pulsar.client.api.MessageId
import org.apache.pulsar.client.impl.MessageIdImpl

val topic = "my-topic"
val msgId: MessageId = new MessageIdImpl(ledgerId, entryId, partitionIndex)
val startOffsets = JsonUtils.topicOffsets(Map(topic -> msgId))

query = spark.readStream
  .format("pulsar")
  .option("service.url", "pulsar://broker.example.com:6650")
  .option("topic", topic)
  .option("startingOffsets", startOffsets)
  .load()