read_pubsub streaming funkce s návratovou hodnotou tabulky

Platí pro:zaškrtnuto ano Databricks SQL zaškrtnuto ano Databricks Runtime 13.3 LTS a vyšší verze

Vrátí tabulku se záznamy načtenými z Pub/Sub z daného tématu. Podporuje pouze dotazy streamování.

Syntaxe

read_pubsub( { parameter => value } [, ...])

Argumenty

read_pubsub vyžaduje vyvolání pojmenovaného parametru.

Jedinými povinnými argumenty jsou subscriptionId, projectIda topicId. Všechny ostatní argumenty jsou volitelné.

Úplné popisy argumentů najdete v tématu Konfigurace možností čtení Pub/Sub streamingu.

Databricks doporučuje používat tajné kódy při poskytování možností autorizace. Viz secret funkce.

Podrobnosti o konfiguraci přístupu k pub/sub naleznete v tématu Konfigurace přístupu k Pub/Sub.

Parametr Typ Popis
subscriptionId STRING Povinný identifikátor přiřazený k předplatnému Pub/Sub.
projectId STRING Povinné je ID projektu Google Cloud přidružené k tématu Pub/Sub.
topicId STRING Povinné, ID nebo název tématu Pub/Sub pro odběr.
clientEmail STRING E-mailová adresa přidružená k účtu služby pro ověřování.
clientId STRING ID klienta přidružené k účtu služby pro ověřování.
privateKeyId STRING ID privátního klíče přidruženého k účtu služby.
privateKey STRING Privátní klíč přidružený k účtu služby pro ověřování.

Tyto argumenty se používají k dalšímu vyladění při čtení z Pub/Sub:

Parametr Typ Popis
numFetchPartitions STRING Volitelné s výchozím počtem exekutorů. Počet paralelních úloh Spark, které načítají záznamy z předplatného.
deleteSubscriptionOnStreamStop BOOLEAN Volitelné s výchozím nastavením false. Je-li nastavena hodnota true, odběr předaný streamu se odstraní po ukončení úlohy streamování.
maxBytesPerTrigger STRING Měkký limit velikosti dávky, která se má zpracovat během každé aktivované mikrodávkové dávky. Výchozí hodnota je žádná.
maxRecordsPerFetch STRING Počet záznamů, které se mají načíst před zpracováním záznamů na každou úlohu. Výchozí hodnota je 1000.
maxFetchPeriod STRING Doba trvání pro načtení každého úkolu před zpracováním záznamů. Výchozí hodnota je 10s.

Návraty

Tabulka záznamů Pub/Sub s následujícím schématem. Sloupec atributů může mít hodnotu null, ale všechny ostatní sloupce nemají hodnotu null.

Název Datový typ Vynulovatelné Standard Popis
messageId STRING Ne Jedinečný identifikátor zprávy Pub/Sub.
payload BINARY Ne Obsah zprávy Pub/Sub
attributes STRING Ano Páry klíčů a hodnot představující atributy zprávy Pub/Sub. Jedná se o řetězec kódovaný ve formátu JSON.
publishTimestampInMillis BIGINT Ne Časové razítko při publikování zprávy v milisekundách.
sequenceNumber BIGINT Ne Jedinečný identifikátor záznamu v rámci jeho fragmentu.

Příklady

-- Streaming Ingestion from Pubsub using Google Service Account secrets
> CREATE STREAMING TABLE testing.streaming_table AS
  SELECT * FROM STREAM read_pubsub (
                subscriptionId => 'app-events-1234',
                projectId => 'app-events-project',
                topicId => 'app-events-topic',
                clientEmail => secret('app-events', 'clientEmail'),
                clientId => secret('app-events', 'clientId'),
        privateKeyId => secret('app-events', 'privateKeyId'),
                privateKey => secret('app-events', 'privateKey')
);

-- A streaming query when a service account is associated with the cluster
> CREATE STREAMING TABLE testing.streaming_table AS
  SELECT * FROM STREAM read_pubsub (
                subscriptionId => 'app-events-1234',
                projectId => 'app-events-project',
                topicId => 'app-events-topic'
);

Data by nyní musela dotázat z testing.streaming_table pro další analýzu.

Chybné dotazy:

-- Missing topicId option
> CREATE STREAMING TABLE testing.streaming_table AS
  SELECT * FROM STREAM read_pubsub (
                subscriptionId => 'app-events-1234',
                projectId => 'app-events-project'
);

-- Limit is too high for an option, MAX_RECORDS_PER_FETCH_LIMIT
> CREATE STREAMING TABLE testing.streaming_table AS
  SELECT * FROM STREAM read_pubsub (
                subscriptionId => 'app-events-1234',
                projectId => 'app-events-project',
                topicId => 'app-events-topic',
                maxRecordsPerFetchLimit => '1000001'
);
  • Last updated on