Abonneren op Google Pub/Sub

Gebruik de ingebouwde connector om u te abonneren op Google Pub/Sub. Deze connector heeft exact-once-verwerkingssemantiek voor rijen afkomstig van de abonnee.

Notitie

Pub/Sub levert mogelijk dubbele rijen, of rijen kunnen in een andere volgorde bij de abonnee aankomen. U moet code schrijven om dubbele en verouderde rijen af te handelen.

Een Pub/Substream configureren

In het volgende codevoorbeeld ziet u hoe u een Structured Streaming-leesbewerking configureert vanuit Pub/Sub en verifieert met persoonlijke sleutels.

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

Voor meer configuratieopties, zie Opties configureren voor Pub/Sub-streamingleesbewerkingen.

Toegang tot Pub/Sub configureren

Uw inloggegevens moeten de volgende rollen hebben:

Rollen	Vereist of optioneel	Hoe rol wordt gebruikt
roles/pubsub.viewer of roles/viewer	Vereist	Controleert of het abonnement bestaat en het abonnement ophaalt.
roles/pubsub.subscriber	Vereist	Hiermee worden gegevens opgehaald uit een abonnement.
roles/pubsub.editor of roles/editor	Optioneel	Hiermee kunt u een abonnement aanmaken als er nog geen bestaat en kunt u het deleteSubscriptionOnStreamStop gebruiken om abonnementen te verwijderen bij het beëindigen van een stream.

Notitie

Als u roles/pubsub.viewer en roles/pubsub.subscriber op resourceniveau toekent in plaats van op projectniveau, moet u beide rollen toepassen op zowel het topic als het abonnement. Als u de optionele rollen roles/pubsub.editor of roles/editor niet gebruikt, is het niet voldoende om de vereiste rollen alleen voor het onderwerp toe te kennen.

Databricks raadt u aan geheimen te gebruiken bij het gebruik van sleutels. De volgende opties zijn vereist om een verbinding te autoriseren:

• clientEmail
• clientId
• privateKey
• privateKeyId

Inzicht in het pub-/subschema

Het schema voor de stream komt overeen met de rijen die worden opgehaald uit Pub/Sub, zoals beschreven in de volgende tabel:

Veld	Typologie
messageId	StringType
payload	ArrayType[ByteType]
attributes	StringType
publishTimestampInMillis	LongType

Opties configureren voor Pub/Sub-streaminglezen

Bepaalde Pub/Sub-configuratieopties gebruiken het concept van ophalen in plaats van microbatches. Dit is een intern implementatiedetail en de opties werken vergelijkbaar met andere Structured Streaming-connectors, behalve dat rijen worden opgehaald en vervolgens verwerkt.

Zie Pub/Sub voor de volledige lijst met opties.

Incrementele batchverwerking gebruiken met Pub/Sub

U kunt Trigger.AvailableNow gebruiken om beschikbare rijen uit de Pub/Sub-bronnen te verwerken als een incrementele batch.

Azure Databricks registreert de tijdstempel wanneer u begint met lezen met de Trigger.AvailableNow instelling. Rijen die door de batch worden verwerkt, bevatten alle eerder opgehaalde gegevens en nieuwe gepubliceerde rijen met een tijdstempel die kleiner is dan de vastgelegde begintijdstempel. Zie AvailableNowvoor meer informatie: Incrementele batchverwerking.

Pub/Sub-streamingmetrieken bewaken

Metrische gegevens over de voortgang van gestructureerd streamen rapporteren het aantal rijen dat is opgehaald en gereed is voor verwerking, de grootte van de opgehaalde en gereed voor verwerking van rijen en het aantal duplicaten dat is gezien sinds het begin van de stream.

Hier volgt een voorbeeld van metrische gegevens van Pub/Sub:

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

Beperkingen

Pub/Sub biedt geen ondersteuning voor speculatieve uitvoering met spark.speculation.