Membaca dari Cloud Storage ke Dataflow

Untuk membaca data dari Cloud Storage ke Dataflow, gunakan konektor I/O TextIO atau AvroIO Apache Beam.

Menyertakan dependensi library Google Cloud

Untuk menggunakan konektor TextIO atau AvroIO dengan Cloud Storage, sertakan dependensi berikut. Library ini menyediakan handler skema untuk nama file "gs://".

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

apache-beam[gcp]==VERSION

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

Untuk mengetahui informasi selengkapnya, lihat Menginstal Apache Beam SDK.

Mengaktifkan konektor I/O gRPC di Apache Beam pada Dataflow

Anda dapat terhubung ke Cloud Storage menggunakan gRPC melalui konektor I/O Apache Beam di Dataflow. gRPC adalah framework panggilan prosedur jarak jauh (RPC) open source berperforma tinggi yang dikembangkan oleh Google yang dapat Anda gunakan untuk berinteraksi dengan Cloud Storage.

Untuk mempercepat permintaan baca tugas Dataflow Anda ke Cloud Storage, Anda dapat mengaktifkan konektor I/O Apache Beam di Dataflow untuk menggunakan gRPC.

Anda dapat mengonfigurasi konektor I/O Apache Beam di Dataflow untuk menghasilkan metrik terkait gRPC di Cloud Monitoring. Metrik terkait gRPC dapat membantu Anda melakukan hal berikut:

• Pantau dan optimalkan performa permintaan gRPC ke Cloud Storage.
• Memecahkan masalah dan melakukan debug.
• Dapatkan insight tentang penggunaan dan perilaku aplikasi Anda.

Keparalelan

Konektor TextIO dan AvroIO mendukung dua tingkat paralelisme:

• Setiap file diberi kunci secara terpisah, sehingga beberapa pekerja dapat membacanya.
• Jika file tidak dikompresi, konektor dapat membaca sub-rentang setiap file secara terpisah, sehingga menghasilkan tingkat paralelisme yang sangat tinggi. Pemecahan ini hanya mungkin dilakukan jika setiap baris dalam file adalah kumpulan data yang bermakna. Misalnya, fitur ini tidak tersedia secara default untuk file JSON.

Performa

Tabel berikut menunjukkan metrik performa untuk membaca dari Cloud Storage. Beban kerja dijalankan di satu pekerja e2-standard2, menggunakan Apache Beam SDK 2.49.0 untuk Java. Mereka tidak menggunakan Runner v2.

Metrik ini didasarkan pada pipeline batch sederhana. Benchmark ini dimaksudkan untuk membandingkan performa antara konektor I/O, dan tidak selalu mewakili pipeline dunia nyata. Performa pipeline Dataflow bersifat kompleks, dan merupakan fungsi dari jenis VM, data yang diproses, performa sumber dan sink eksternal, serta kode pengguna. Metrik didasarkan pada menjalankan Java SDK, dan tidak mewakili karakteristik performa SDK bahasa lainnya. Untuk mengetahui informasi selengkapnya, lihat Performa IO Beam.

Praktik terbaik

• Hindari penggunaan watchForNewFiles dengan Cloud Storage. Pendekatan ini tidak dapat diskalakan dengan baik untuk pipeline produksi besar, karena konektor harus menyimpan daftar file yang telah dilihat dalam memori. Daftar tidak dapat dihapus dari memori, yang mengurangi memori kerja pekerja dari waktu ke waktu. Pertimbangkan untuk menggunakan notifikasi Pub/Sub untuk Cloud Storage sebagai gantinya. Untuk mengetahui informasi selengkapnya, lihat Pola pemrosesan file.

• Jika nama file dan konten file adalah data yang berguna, gunakan class FileIO untuk membaca nama file. Misalnya, nama file dapat berisi metadata yang berguna saat memproses data dalam file. Untuk informasi selengkapnya, lihat Mengakses nama file. Dokumentasi FileIO juga menunjukkan contoh pola ini.

Contoh

Contoh berikut menunjukkan cara membaca dari Cloud Storage.

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;
  }
}

Langkah berikutnya

• Baca dokumentasi TextIO API.
• Lihat daftar template yang disediakan Google.