Dateien aus einer PCollection von GCS-Dateinamen in Pipeline lesen?

Question

Ich habe eine Streaming-Pipeline angeschlossen an Pub/Sub, die Dateinamen von GCS-Dateien veröffentlicht. Von dort möchte ich jede Datei lesen und die Ereignisse in jeder Zeile analysieren (die Ereignisse sind, was ich letztendlich verarbeiten möchte).

Kann ich TextIO verwenden? Können Sie es in einer Streaming-Pipeline verwenden, wenn der Dateiname während der Ausführung definiert wird (im Gegensatz zur Verwendung von TextIO als Quelle und der Datei (en) sind bei der Konstruktion bekannt)? Wenn nicht ich denke an folgend etwas wie das tun:

das Thema aus pub/sub Pardo Erhalten Sie jede Datei zu lesen und die Linien Prozess der Zeilen der Datei bekommen ...

Kann ich Verwenden Sie den FileBasedReader oder etwas ähnliches in diesem Fall, um die Dateien zu lesen? Die Dateien sind nicht zu groß, so dass ich nicht das Lesen einer einzelnen Datei parallelisieren müsste, aber ich müsste viele Dateien lesen.

Answer 1

Sie können die TextIO.readAll() Transformation verwenden, die kürzlich zu Beam in #3443 hinzugefügt wurde. Beispiel:

PCollection<String> filenames = p.apply(PubsubIO.readStrings()...); 
PCollection<String> lines = filenames.apply(TextIO.readAll()); 

Dies wird alle Zeilen in jeder Datei lesen, die über Pubsub ankommen.