read_pulsar streaming funkce s návratovou hodnotou tabulky

Platí pro:označeno zaškrtnutím ano Databricks SQL označeno zaškrtnutím ano Databricks Runtime 14.1 a vyšší

Důležité

Tato funkce je ve verzi Public Preview.

Vrátí tabulku se záznamy načtenými z Pulsaru.

Tato funkce s hodnotou tabulky podporuje pouze streamování, nikoli dávkové dotazy.

Syntaxe

read_pulsar ( { option_key => option_value } [, ...] )

Argumenty

Tato funkce vyžaduje vyvolání pojmenovaného parametru pro klíče možností.

Možnosti serviceUrl a topic jsou povinné.

Popisy argumentů jsou zde stručné. Další popisy najdete v dokumentaci ke strukturovanému streamování Pulsear .

Možnost Typ Výchozí Popis
adresa služby STRING Povinné Identifikátor URI služby Pulsear.
téma STRING Povinné Téma, o kterém se má číst.
předdefinované předplatné STRING Nic Předdefinovaný název předplatného používaný konektorem ke sledování průběhu aplikace Spark.
PředponaPředplatného STRING Nic Předpona používaná konektorem ke generování náhodného předplatného ke sledování průběhu aplikace Spark.
pollTimeoutMs DLOUHÝ 120000 Časový limit pro čtení zpráv z Pulsaru v milisekundách.
failOnDataLoss BOOLEOVSKÝ pravda Určuje, jestli se dotaz nezdaří, když dojde ke ztrátě dat (například témata se odstraní nebo zprávy se odstraní kvůli zásadám uchovávání informací).
startingOffsets STRING nejnovější Počáteční bod při spuštění dotazu, buď nejstaršího, nejnovějšího, nebo řetězce JSON, který určuje konkrétní posun. Pokud je nejnovější, přečte si čtenář nejnovější záznamy po spuštění. Pokud je nejstarší, čtečka čte od nejranějšího možného místa. Uživatel může také zadat řetězec JSON, který určuje konkrétní posun.
počáteční čas STRING Nic Při zadání bude zdroj Pulsear číst zprávy od pozice zadaného počátečního času.

Pro ověřování pulzárního klienta se používají následující argumenty:

Možnost Typ Výchozí Popis
pulsearClientAuthPluginClassName STRING Nic Název ověřovacího modulu.
pulsarClientAuthParams STRING Nic Parametry ověřovacího modulu plug-in.
pulsarClientUseKeyStoreTls STRING Nic Určuje, jestli se má úložiště klíčů používat pro ověřování tls.
pulsarClientTlsTrustStoreType STRING Nic Typ souboru TrustStore pro ověřování tls.
pulsarClientTlsTrustStorePath STRING Nic Cesta k souboru TrustStore pro ověřování TLS.
pulsearClientTlsTrustStorePassword STRING Nic Heslo TrustStore pro ověřování tls

Tyto argumenty se používají pro konfiguraci a ověřování řízení přístupu Pulsar, konfigurace Pulsar Admin se vyžaduje pouze v případě, že je povoleno řízení přístupu (pokud je nastaven maxBytesPerTrigger).

Možnost Typ Výchozí Popis
maxBytesPerTrigger BIGINT Nic Měkký limit maximálního počtu bajtů, které chceme zpracovat na mikrobatch. Pokud je zadáno, musí být zadána také adresa admin.url.
adminUrl STRING Nic Konfigurace serviceHttpUrl Pulsar. Vyžaduje se pouze v případě, že je zadán maxBytesPerTrigger.
pulsearAdminAuthPlugin STRING Nic Název ověřovacího modulu.
pulsarAdminAuthParams STRING Nic Parametry ověřovacího modulu plug-in.
pulsarClientUseKeyStoreTls STRING Nic Určuje, jestli se má úložiště klíčů používat pro ověřování tls.
pulsarAdminTlsTrustStoreType STRING Nic Typ souboru TrustStore pro ověřování tls.
pulsarAdminTlsTrustStorePath STRING Nic Cesta k souboru TrustStore pro ověřování TLS.
pulsarAdminTlsTrustStorePassword STRING Nic Heslo TrustStore pro ověřování tls

Návraty

Tabulka pulsárních záznamů s následujícím schématem.

  • __key STRING NOT NULL: Klíč zprávy Pulsar.

  • value BINARY NOT NULL: Hodnota zprávy Pulsar.

    Poznámka: Pro témata se schématem Avro nebo JSON se místo načítání obsahu do pole binární hodnoty obsah rozšíří, aby se zachovaly názvy polí a typy polí tématu Pulsear.

  • __topic STRING NOT NULL: Název tématu Pulsar.

  • __messageId BINARY NOT NULL: Identifikátor zprávy Pulsar.

  • __publishTime TIMESTAMP NOT NULL: Čas publikace zprávy Pulsar.

  • __eventTime TIMESTAMP NOT NULL: Čas události pulsarové zprávy.

  • __messageProperties MAP<STRING, STRING>: Vlastnosti pulzární zprávy.

Příklady

-- Streaming from Pulsar
> CREATE STREAMING TABLE testing.streaming_table AS
  SELECT * FROM STREAM read_pulsar(
      serviceUrl => 'pulsar://broker.example.com:6650',
      startingOffsets => 'earliest',
      topic => 'my-topic');

-- Streaming Ingestion from Pulsar with authentication
> CREATE STREAMING TABLE testing.streaming_table AS
  SELECT * FROM STREAM read_pulsar(
        serviceUrl => 'pulsar://broker.example.com:6650',
        startingOffsets => 'earliest',
        topic => 'my-topic',
        pulsarClientAuthPluginClassName => 'org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls',
        pulsarClientAuthParams => 'keyStoreType:JKS,keyStorePath:/var/private/tls/client.keystore.jks,keyStorePassword:clientpw'
        );

The data can now to be queried from the testing.streaming_table for further analysis.
  • Last updated on