Přihlášení k odběru Google Pub/Sub

Pomocí integrovaného konektoru se můžete přihlásit k odběru Google Pub/Sub. Tento konektor má přesně jedno zpracování sémantiky pro řádky od odběratele.

Poznámka:

Pub/Sub může publikovat duplicitní řádky nebo mohou řádky dorazit odběrateli v nesprávném pořadí. Je nutné napsat kód pro zpracování duplicitních a zastaralých řádků.

Konfigurace streamu Pub/Sub

Následující příklad kódu ukazuje, jak nakonfigurovat strukturované streamování přečtené z pub/Sub a ověřit pomocí privátních klíčů.

Python

auth_options = {
    "clientId": client_id,
    "clientEmail": client_email,
    "privateKey": private_key,
    "privateKeyId": private_key_id
}

query = (spark.readStream
  .format("pubsub")
  .option("subscriptionId", "mysub")
  .option("topicId", "mytopic")
  .option("projectId", "myproject")
  .options(auth_options)
  .load()
)

Scala

val authOptions: Map[String, String] =
  Map("clientId" -> clientId,
      "clientEmail" -> clientEmail,
      "privateKey" -> privateKey,
      "privateKeyId" -> privateKeyId)

val query = spark.readStream
  .format("pubsub")
  // Creates a Pub/Sub subscription if one does not already exist with this ID
  .option("subscriptionId", "mysub")
  .option("topicId", "mytopic")
  .option("projectId", "myproject")
  .options(authOptions)
  .load()

SQL

CREATE OR REFRESH STREAMING TABLE pubsub_raw
AS SELECT * FROM STREAM read_pubsub(
  subscriptionId => 'mysub',
  projectId => 'myproject',
  topicId => 'mytopic',
  clientEmail => secret('pubsub-scope', 'clientEmail'),
  clientId => secret('pubsub-scope', 'clientId'),
  privateKeyId => secret('pubsub-scope', 'privateKeyId'),
  privateKey => secret('pubsub-scope', 'privateKey')
);

Další možnosti konfigurace najdete v tématu Nastavení konfigurace pro Pub/Sub čtení streamu.

Konfigurace přístupu k pub/Sub

Vaše přihlašovací údaje musí mít následující role:

Role	Požadované nebo volitelné	Jak se role používá
roles/pubsub.viewer nebo roles/viewer	Požaduje se	Zkontroluje, jestli existuje předplatné a získá předplatné.
roles/pubsub.subscriber	Požaduje se	Načte data z předplatného.
roles/pubsub.editor nebo roles/editor	Volitelné	Povolí vytvoření předplatného, pokud neexistuje, a umožní použití deleteSubscriptionOnStreamStop k odstranění předplatných při ukončení streamu.

Poznámka:

Pokud přiřadíte roles/pubsub.viewer a roles/pubsub.subscriber na úrovni zdroje namísto na úrovni projektu, musíte obě role přiřadit jak k tématu, tak k předplatnému. Pokud nepoužíváte volitelné role roles/pubsub.editor nebo roles/editor, nestačí udělit požadované role pouze pro dané téma.

Databricks doporučuje používat tajné kódy při použití klíčů. K autorizaci připojení se vyžadují následující možnosti:

• clientEmail
• clientId
• privateKey
• privateKeyId

Vysvětlení schématu Pub/Sub

Schéma datového proudu odpovídá řádkům načteným z pub/Sub, jak je popsáno v následující tabulce:

Pole	Typ
messageId	StringType
payload	ArrayType[ByteType]
attributes	StringType
publishTimestampInMillis	LongType

Konfigurace možností streamování čtení v Pub/Sub

Některé možnosti konfigurace Pub/Sub používají místo mikrodávek koncept načítání. Jedná se o interní podrobnosti implementace a možnosti fungují podobně jako jiné konektory strukturovaného streamování s tím rozdílem, že řádky se načítají a pak zpracovávají.

Úplný seznam možností najdete v tématu Pub/Sub.

Použití přírůstkového dávkového zpracování s Pub/Sub

Pomocí Trigger.AvailableNow můžete zpracovat dostupné řádky ze zdrojů Pub/Sub ve formě přírůstkové dávky.

Azure Databricks zaznamenává časové razítko, když zahájíte čtení s nastavením Trigger.AvailableNow. Řádky zpracovávané dávkou zahrnují všechna dříve načtená data a všechny nově publikované řádky s časovým razítkem kratším než zaznamenané počáteční časové razítko. Další informace naleznete v tématu AvailableNow: Přírůstkové dávkové zpracování.

Monitorování streamovacích metrik Pub/Sub

Metriky průběhu strukturovaného streamování hlásí počet řádků načtených a připravených ke zpracování, velikost řádků načtených a připravených ke zpracování a počet duplicit zobrazených od spuštění datového proudu.

Následuje příklad metrik Pub/Sub:

"metrics" : {
  "numDuplicatesSinceStreamStart" : "1",
  "numRecordsReadyToProcess" : "1",
  "sizeOfRecordsReadyToProcess" : "8"
}

Omezení

Pub/Sub nepodporuje spekulativní spuštění s spark.speculation.