Chaînage d'une autre transformation après DataStoreIO.Write

Question

Je crée un pipeline de flux de données Google à l'aide d'Apache Beam Java SDK. J'ai quelques transformations là, et je crée enfin une collection d'Entités (PCollection < Entity>). J'ai besoin d'écrire cela dans le Google DataStore et puis, effectuer une autre transformation APRÈS que toutes les entités ont été écrites. (comme la diffusion des ID des objets enregistrés via un message PubSub à plusieurs abonnés).

Maintenant, la façon de stocker un PCollection est par:.. entities.DatastoreIO.v1() écrire() withProjectId ("abc")

Ce retourne un objet PDone, et je ne sais pas comment Je peux enchaîner une autre transformation pour se produire après que cette écriture() soit terminée. Étant donné que l'appel DatastoreIO.write() ne renvoie pas un PCollection, je ne suis pas en mesure de poursuivre le pipeline. J'ai 2 questions:

1. Comment puis-je obtenir les ID des objets écrits dans le magasin de données?

2. Comment est-ce que je peux attacher une autre transformation qui agira après que toutes les entités aient été sauvegardées?

Answer 1

Nous n'avons pas une bonne façon de le faire ou l'autre de ces choses (retour ID d'entités de magasin de données écrites, ou attendre que les entités ont été écrites), bien que ce soit loin de la première demande similaire (les gens ont a demandé cela pour BigQuery, par exemple) et nous y pensons. À l'heure actuelle, votre seule option consiste à attendre que le pipeline entier se termine, par ex. via pipeline.run().waitUntilFinish(), puis en faisant ce que vous vouliez dans votre programme principal (par exemple, vous pouvez exécuter un autre pipeline).