Abonnieren Sie Google Pub/Sub

Verwenden Sie den integrierten Connector, um Google Pub/Sub zu abonnieren. Dieser Connector verfügt über eine Exactly-Once-Verarbeitungssemantik für Zeilen vom Abonnenten.

Hinweis

Pub/Sub veröffentlicht möglicherweise doppelte Zeilen, oder Zeilen kommen möglicherweise außerhalb der Reihenfolge an den Abonnenten. Sie müssen Code schreiben, um doppelte und nicht sortierte Zeilen zu behandeln.

Konfigurieren eines Pub/Sub-Streams

Im folgenden Codebeispiel wird gezeigt, wie Sie einen strukturierten Streaming-Lesevorgang aus Pub/Sub konfigurieren und sich mit privaten Schlüsseln authentifizieren.

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')
);

Weitere Konfigurationsoptionen finden Sie unter Konfigurationsoptionen für Pub/Sub-Streaming-Lesen.

Konfigurieren des Zugriffs auf Pub/Sub

Ihre Anmeldeinformationen müssen über die folgenden Rollen verfügen:

Rollen	Erforderlich oder optional	Wie die Rolle verwendet wird
roles/pubsub.viewer oder roles/viewer	Erforderlich	Überprüft, ob das Abonnement vorhanden ist, und ruft das Abonnement ab.
roles/pubsub.subscriber	Erforderlich	Ruft Daten aus einem Abonnement ab.
roles/pubsub.editor oder roles/editor	Wahlfrei	Ermöglicht die Erstellung eines Abonnements, wenn keines vorhanden ist, und ermöglicht die Verwendung der deleteSubscriptionOnStreamStop zum Löschen von Abonnements bei der Stream-Beendigung.

Hinweis

Wenn Sie roles/pubsub.viewer und roles/pubsub.subscriber auf Ressourcenebene statt auf Projektebene zuweisen, müssen Sie beide Rollen sowohl auf das Thema als auch auf das Abonnement anwenden. Wenn Sie die optionalen Rollen roles/pubsub.editor oder roles/editor nicht verwenden, reicht es nicht aus, die erforderlichen Rollen nur für das Thema zuzuweisen.

Databricks empfiehlt, geheime Schlüssel bei der Verwendung von Schlüsseln zu verwenden. Die folgenden Optionen sind erforderlich, um eine Verbindung zu autorisieren:

• clientEmail
• clientId
• privateKey
• privateKeyId

Grundlegendes zum Pub/Sub-Schema

Das Schema für den Datenstrom entspricht den Zeilen, die aus Pub/Sub abgerufen werden, wie in der folgenden Tabelle beschrieben:

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

Konfigurieren von Optionen für den Pub/Sub-Streaming-Lesevorgang

Einige Pub/Sub-Konfigurationsoptionen verwenden das Konzept der Abrufe anstelle von Mikrobatches. Dies ist ein internes Implementierungsdetail, und die Optionen funktionieren ähnlich wie andere Strukturierte Streaming-Connectors, mit der Ausnahme, dass Zeilen abgerufen und dann verarbeitet werden.

Die vollständige Liste der Optionen finden Sie unter Pub/Sub.

Verwenden der inkrementellen Batchverarbeitung mit Pub/Sub

Sie können Trigger.AvailableNow verwenden, um verfügbare Zeilen aus den Pub/Sub-Quellen als inkrementellen Batch zu nutzen.

Azure Databricks zeichnet den Zeitstempel auf, wenn Sie einen Lesevorgang mit der Trigger.AvailableNow Einstellung beginnen. Vom Batch verarbeitete Zeilen umfassen alle zuvor abgerufenen Daten und alle neu veröffentlichten Zeilen mit einem Zeitstempel, der kleiner als der aufgezeichnete Startzeitstempel ist. Weitere Informationen finden Sie unter AvailableNow: Inkrementelle Batchverarbeitung.

Überwachen von Pub/Sub Streaming-Metriken

Metriken für strukturiertes Streaming melden die Anzahl der abgerufenen und verarbeiteten Zeilen, die Größe der abgerufenen und verarbeiteten Zeilen sowie die Anzahl der Duplikate, die seit dem Start des Datenstroms angezeigt werden.

Nachfolgend sehen Sie ein Beispiel für Pub/Sub-Metriken:

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

Begrenzungen

Pub/Sub unterstützt keine spekulative Ausführung mit spark.speculation.