Udostępnij za pośrednictwem

funkcja read_pulsar zwracająca wartości tabeli strumieniowo

  • Artykuł

Dotyczy:zaznacz pole wyboru oznaczone jako tak Databricks SQL zaznacz pole wyboru oznaczone jako tak Databricks Runtime 14.1 i nowsze

Ważne

Ta funkcja jest dostępna w publicznej wersji zapoznawczej.

Zwraca tabelę z rekordami odczytanymi z pulsaru.

Ta funkcja tabelaryczna obsługuje tylko przesyłanie strumieniowe i nie obsługuje zapytań wsadowych.

Składnia

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

Argumenty

Ta funkcja wymaga wywołania nazwanego parametru dla kluczy opcji.

Opcje serviceUrl i topic są obowiązkowe.

Opisy argumentów są krótkie tutaj. Aby uzyskać rozszerzone opisy, sprawdź dokumentację Pulsar dotyczącą ustrukturyzowanego przesyłania strumieniowego.

Opcja Typ Domyślny opis
serviceUrl STRUNA Obowiązkowy Identyfikator URI usługi Pulsar.
temat STRUNA Obowiązkowy Temat do odczytania.
wstępnie zdefiniowana subskrypcja STRUNA Brak Wstępnie zdefiniowana nazwa subskrypcji używana przez łącznik do śledzenia postępu aplikacji Spark.
przedrostek subskrypcji STRUNA Brak Prefiks używany przez łącznik do generowania losowej subskrypcji do śledzenia postępu aplikacji platformy Spark.
pollTimeoutMs DŁUGI 120000 Limit czasu odczytywania komunikatów z Pulsar w milisekundach.
błądPrzyUtracieDanych BOOLEAN prawda Określa, czy zapytanie nie powiodło się w przypadku utraty danych (na przykład tematy są usuwane lub komunikaty są usuwane z powodu zasad przechowywania).
wartości początkowe STRUNA latest Początek, kiedy zapytanie jest uruchamiane, może być zdefiniowany jako najwcześniejszy, najnowszy lub jako ciąg JSON określający określone przesunięcie. Jeśli opcja "najnowsze" jest włączona, czytnik odczytuje najnowsze rekordy po rozpoczęciu pracy. Jeśli to jest możliwe najwcześniej, czytnik odczytuje od najwcześniejszego przesunięcia. Użytkownik może również określić ciąg JSON, który określa określone przesunięcie.
czas rozpoczęcia STRUNA Brak Po określeniu źródło Pulsar odczytuje komunikaty, zaczynając od pozycji oznaczonej określonym czasem rozpoczęcia.

Następujące argumenty są używane do uwierzytelniania klienta pulsar:

Opcja Typ Domyślny opis
pulsarClientAuthPluginClassName STRUNA Brak Nazwa wtyczki uwierzytelniania.
pulsarClientAuthParams STRUNA Brak Parametry wtyczki uwierzytelniania.
pulsarClientUseKeyStoreTls STRUNA Brak Czy używać magazynu kluczy do uwierzytelniania TLS.
pulsarClientTlsTrustStoreType STRUNA Brak Typ pliku TrustStore na potrzeby uwierzytelniania tls.
pulsarClientTlsTrustStorePath STRUNA Brak Ścieżka pliku TrustStore na potrzeby uwierzytelniania TLS.
pulsarClientTlsTrustStorePassword STRUNA Brak Hasło trustStore na potrzeby uwierzytelniania tls.

Te argumenty są używane do konfiguracji i uwierzytelniania kontroli przyjęć Pulsar, konfiguracja administracyjna Pulsar jest wymagana tylko w przypadku włączenia kontroli przyjęć (gdy parametr maxBytesPerTrigger jest ustawiony).

Opcja Typ Domyślny opis
maxBytesPerTrigger BIGINT Brak Miękki limit maksymalnej liczby bajtów, które chcemy przetworzyć na mikropartię. Jeśli jest to określone, należy również określić adres admin.url.
adminUrl STRUNA Brak Konfiguracja usługi Pulsar serviceHttpUrl. Wymagane tylko w przypadku określenia parametru maxBytesPerTrigger.
pulsarAdminAuthPlugin STRUNA Brak Nazwa wtyczki uwierzytelniania.
pulsarAdminAuthParams STRUNA Brak Parametry wtyczki uwierzytelniania.
pulsarClientUseKeyStoreTls SZNUREK Brak Czy używać magazynu kluczy do uwierzytelniania TLS.
pulsarAdminTlsTrustStoreType STRUNA Brak Typ pliku TrustStore na potrzeby uwierzytelniania tls.
pulsarAdminTlsTrustStorePath STRUNA Brak Ścieżka pliku TrustStore do uwierzytelniania TLS.
pulsarAdminTlsTrustStorePassword STRUNA Brak Hasło trustStore na potrzeby uwierzytelniania tls.

Zwraca

Tabela rekordów pulsarnych z następującym schematem.

  • __key STRING NOT NULL: Klucz wiadomości Pulsar.

  • value BINARY NOT NULL: wartość komunikatu Pulsar.

    Uwaga: w przypadku tematów ze schematem Avro lub JSON zamiast ładowania zawartości do pola wartości binarnej zawartość zostanie rozszerzona w celu zachowania nazw pól i typów pól tematu Pulsar.

  • __topic STRING NOT NULL: Nazwa tematu Pulsar.

  • __messageId BINARY NOT NULL: identyfikator komunikatu Pulsar.

  • __publishTime TIMESTAMP NOT NULL: Czas publikowania komunikatu pulsar.

  • __eventTime TIMESTAMP NOT NULL: czas zdarzenia komunikatu Pulsar.

  • __messageProperties MAP<STRING, STRING>: właściwości wiadomości Pulsar.

Przykłady

-- 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.