Membaca dari Cloud Storage ke Dataflow

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

Menyertakan dependensi Google Cloud library

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

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"

Untuk mengetahui informasi selengkapnya, lihat Menginstal Apache Beam SDK.

Mengaktifkan gRPC di konektor I/O Apache Beam di 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 ke Cloud Storage, Anda dapat mengaktifkan konektor I/O Apache Beam di Dataflow untuk menggunakan gRPC.

Command line

  1. Pastikan Anda menggunakan Apache Beam SDK versi 2.55.0 atau yang lebih baru.
  2. Untuk menjalankan tugas Dataflow, gunakan opsi pipeline --additional-experiments=use_grpc_for_gcs. Untuk mengetahui informasi tentang berbagai opsi pipeline, lihat Flag opsional.

Apache Beam SDK

  1. Pastikan Anda menggunakan Apache Beam SDK versi 2.55.0 atau yang lebih baru.
  2. Untuk menjalankan tugas Dataflow, gunakan --experiments=use_grpc_for_gcs opsi pipeline. Untuk mengetahui informasi tentang berbagai opsi pipeline, lihat Opsi dasar.

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:

  • Memantau dan mengoptimalkan performa permintaan gRPC ke Cloud Storage.
  • Memecahkan masalah dan men-debug masalah.
  • Mendapatkan insight tentang penggunaan dan perilaku aplikasi Anda.

Untuk mengetahui informasi tentang cara mengonfigurasi konektor I/O Apache Beam di Dataflow untuk menghasilkan metrik terkait gRPC, lihat Menggunakan metrik sisi klien. Jika pengumpulan metrik tidak diperlukan untuk kasus penggunaan Anda, Anda dapat memilih untuk tidak ikut pengumpulan metrik. Untuk mengetahui petunjuknya, lihat Memilih tidak ikut metrik sisi klien.

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. Pembuatan sub-rentang ini hanya dapat dilakukan jika setiap baris dalam file adalah rekaman yang bermakna. Misalnya, fitur ini tidak tersedia secara default untuk file JSON.

Performa

Tabel berikut menunjukkan metrik performa untuk membaca dari Cloud Storage. Workload dijalankan pada satu pekerja e2-standard2, menggunakan Apache Beam SDK 2.49.0 untuk Java. Workload tersebut tidak menggunakan Runner v2.

100 M rekaman | 1 kB | 1 kolom Throughput (byte) Throughput (elemen)
Membaca 320 MBps 320.000 elemen per detik

Metrik ini didasarkan pada pipeline batch sederhana. Metrik 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 Beam IO.

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 terlihat dalam memori. Daftar tersebut 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 mungkin berisi metadata yang berguna saat memproses data dalam file. Untuk mengetahui informasi selengkapnya, lihat Mengakses nama file. Dokumentasi FileIO juga menunjukkan contoh pola ini.

Contoh

Contoh berikut menunjukkan cara membaca dari Cloud Storage.

Java

Untuk melakukan autentikasi ke Dataflow, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal. 

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