Von Cloud Storage nach Dataflow lesen

Verwenden Sie einen der E/A-Connectors TextIO oder AvroIO von Apache Beam, um Daten aus Cloud Storage in Dataflow zu lesen.

Google Cloud-Bibliotheksabhängigkeit einschließen

Schließen Sie die folgende Abhängigkeit ein, um den TextIO- oder AvroIO-Connector mit Cloud Storage zu verwenden. Diese Bibliothek bietet einen Schema-Handler für "gs://"-Dateinamen.

Java 

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
  <version>${beam.version}</version>
</dependency>

Python 

apache-beam[gcp]==VERSION

Go

import _ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/gcs"

Weitere Informationen finden Sie unter Apache Beam SDK installieren.

gRPC für den Apache Beam-E/A-Connector in Dataflow aktivieren

Sie können über den Apache Beam-E/A-Connector in Dataflow mit gRPC eine Verbindung zu Cloud Storage herstellen. gRPC ist ein leistungsstarkes Open-Source-RPC-Framework (Remote Procedure Call) von Google, mit dem Sie mit Cloud Storage interagieren können.

Um die Leseanfragen Ihres Dataflow-Jobs an Cloud Storage zu beschleunigen, können Sie den Apache Beam-E/A-Connector in Dataflow für die Verwendung von gRPC aktivieren.

Befehlszeile

  1. Verwenden Sie das Apache Beam SDK in der Version 2.55.0 oder höher.
  2. Verwenden Sie die Pipelineoption --additional-experiments=use_grpc_for_gcs, um einen Dataflow-Job auszuführen. Informationen zu den verschiedenen Pipelineoptionen finden Sie unter Optionale Flags.

Apache Beam SDK

  1. Verwenden Sie das Apache Beam SDK in der Version 2.55.0 oder höher.
  2. Verwenden Sie die Pipelineoption --experiments=use_grpc_for_gcs, um einen Dataflow-Job auszuführen. Informationen zu den verschiedenen Pipelineoptionen finden Sie unter Grundlegende Optionen.

Sie können den Apache Beam-E/A-Connector in Dataflow konfigurieren, um gRPC-bezogene Messwerte in Cloud Monitoring zu generieren. Mit den gRPC-bezogenen Messwerten können Sie Folgendes tun:

  • Leistung von gRPC-Anfragen an Cloud Storage überwachen und optimieren
  • Probleme beheben
  • Sie erhalten Informationen zur Nutzung und zum Verhalten Ihrer Anwendung.

Informationen zum Konfigurieren des Apache Beam-E/A-Connectors in Dataflow zum Generieren von gRPC-bezogenen Messwerten finden Sie unter Clientseitige Messwerte verwenden. Wenn die Erfassung von Messwerten für Ihren Anwendungsfall nicht erforderlich ist, können Sie die Erfassung deaktivieren. Eine Anleitung finden Sie unter Clientseitige Messwerte deaktivieren.

Parallelität

Die TextIO- und AvroIO-Connectors unterstützen zwei Parallelisierungsebenen:

  • Einzelne Dateien werden separat verschlüsselt, damit sie von mehreren Workern gelesen werden können.
  • Wenn die Dateien nicht komprimiert sind, kann der Connector Teilbereiche der einzelnen Dateien separat lesen, was zu einer sehr hohen Parallelität führt. Diese Aufteilung ist nur möglich, wenn jede Zeile in der Datei einen sinnvollen Datensatz enthält. So ist er beispielsweise standardmäßig nicht für JSON-Dateien verfügbar.

Leistung

Die folgende Tabelle enthält Leistungsmesswerte für das Lesen aus Cloud Storage. Die Arbeitslasten wurden auf einem e2-standard2-Worker mit dem Apache Beam SDK 2.49.0 für Java ausgeführt. Runner v2 wurde nicht verwendet.

100 Mio. Datensätze | 1 KB | 1 Spalte Durchsatz (Byte) Durchsatz (Elemente)
Lesen 320 Mbit/s 320.000 Elemente pro Sekunde

Diese Messwerte basieren auf einfachen Batch-Pipelines. Sie dienen zum Vergleich der Leistung zwischen E/A-Anschlüssen und sind nicht unbedingt repräsentativ für reale Pipelines. Die Leistung der Dataflow-Pipeline ist komplex und eine Funktion des VM-Typs, der verarbeiteten Daten, der Leistung externer Quellen und Senken sowie des Nutzercodes. Die Messwerte basieren auf der Ausführung des Java SDK und sind nicht repräsentativ für die Leistungsmerkmale anderer Sprach-SDKs. Weitere Informationen finden Sie unter Beam E/A-Leistung.

Best Practices

  • Verwenden Sie watchForNewFiles nicht mit Cloud Storage. Dieser Ansatz skaliert bei großen Produktionspipelines schlecht, da der Connector eine Liste der bereits verarbeiteten Dateien im Arbeitsspeicher speichern muss. Die Liste kann nicht aus dem Arbeitsspeicher gelöscht werden, was den Arbeitsspeicher der Worker im Laufe der Zeit verringert. Verwenden Sie stattdessen Pub/Sub-Benachrichtigungen für Cloud Storage. Weitere Informationen finden Sie unter Muster für die Dateiverarbeitung.

  • Wenn sowohl der Dateiname als auch der Dateiinhalt nützliche Daten sind, verwenden Sie die Klasse FileIO, um Dateinamen zu lesen. Ein Dateiname kann beispielsweise Metadaten enthalten, die bei der Verarbeitung der Daten in der Datei nützlich sind. Weitere Informationen finden Sie unter Auf Dateinamen zugreifen. In der FileIO-Dokumentation finden Sie auch ein Beispiel für dieses Muster.

Beispiel

Im folgenden Beispiel wird gezeigt, wie Daten aus Cloud Storage gelesen werden.

Java

Richten Sie die Standardanmeldedaten für Anwendungen ein, um sich bei Dataflow zu authentifizieren. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten. 

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptors;

public class ReadFromStorage {
  public static Pipeline createPipeline(Options options) {
    var pipeline = Pipeline.create(options);
    pipeline
        // Read from a text file.
        .apply(TextIO.read().from(
            "gs://" + options.getBucket() + "/*.txt"))
        .apply(
            MapElements.into(TypeDescriptors.strings())
                .via(
                    (x -> {
                      System.out.println(x);
                      return x;
                    })));
    return pipeline;
  }
}

Nächste Schritte