Menulis dari Dataflow ke Cloud Storage

Dokumen ini menjelaskan cara menulis data teks dari Dataflow ke Cloud Storage menggunakan konektor I/O TextIO Apache Beam.

Menyertakan dependensi library Google Cloud

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

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 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 dan dapat Anda gunakan untuk berinteraksi dengan Cloud Storage.

Untuk mempercepat permintaan tulis 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 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 opsi pipeline --experiments=use_grpc_for_gcs. Untuk 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:

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

Untuk 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 menggunakan metrik sisi klien.

Keparalelan

Paralelisme ditentukan terutama oleh jumlah shard. Secara default, runner akan otomatis menetapkan nilai ini. Untuk sebagian besar pipeline, sebaiknya gunakan perilaku default. Dalam dokumen ini, lihat Praktik terbaik.

Performa

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

100 juta data | 1 kB | 1 kolom Throughput (byte) Throughput (elemen)
Menulis 130 MBps 130.000 elemen per detik

Metrik ini didasarkan pada pipeline batch sederhana. Pengujian ini dimaksudkan untuk membandingkan performa antara konektor I/O, dan tidak selalu mewakili pipeline di dunia nyata. Performa pipeline Dataflow bersifat kompleks, dan merupakan fungsi dari jenis VM, data yang sedang 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

  • Secara umum, hindari menetapkan jumlah shard tertentu. Hal ini memungkinkan runner memilih nilai yang sesuai untuk skala Anda. Jika Anda menyesuaikan jumlah shard, sebaiknya tulis antara 100 MB dan 1 GB per shard. Namun, nilai optimum mungkin bergantung pada beban kerja.

  • Cloud Storage dapat diskalakan untuk menangani permintaan dalam jumlah sangat besar per detik. Namun, jika pipeline Anda mengalami lonjakan volume operasi tulis yang besar, pertimbangkan untuk menulis ke beberapa bucket, agar tidak membebani satu bucket Cloud Storage untuk sementara.

  • Secara umum, penulisan ke Cloud Storage akan lebih efisien jika setiap penulisan lebih besar (1 kb atau lebih). Menulis data kecil ke sejumlah besar file dapat menghasilkan performa per byte yang lebih buruk.

  • Saat membuat nama file, pertimbangkan untuk menggunakan nama file non-sekuensial, untuk mendistribusikan beban. Untuk informasi selengkapnya, lihat Menggunakan konvensi penamaan yang mendistribusikan beban secara merata di seluruh rentang kunci.

  • Saat memberi nama file, jangan gunakan tanda @ ('@') diikuti dengan angka atau tanda bintang ('*'). Untuk informasi selengkapnya, lihat "@*" dan "@N" adalah spesifikasi sharding yang dicadangkan.

Contoh: Menulis file teks ke Cloud Storage

Contoh berikut membuat pipeline batch yang menulis file teks menggunakan kompresi GZIP:

Java

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

import java.util.Arrays;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.Compression;
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.Create;

public class BatchWriteStorage {
  public interface Options extends PipelineOptions {
    @Description("The Cloud Storage bucket to write to")
    String getBucketName();

    void setBucketName(String value);
  }

  // Write text data to Cloud Storage
  public static void main(String[] args) {
    final List<String> wordsList = Arrays.asList("1", "2", "3", "4");

    var options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    var pipeline = Pipeline.create(options);
    pipeline
        .apply(Create
            .of(wordsList))
        .apply(TextIO
            .write()
            .to(options.getBucketName())
            .withSuffix(".txt")
            .withCompression(Compression.GZIP)
        );
    pipeline.run().waitUntilFinish();
  }
}

Jika PCollection input tidak dibatasi, Anda harus menentukan jendela atau pemicu pada koleksi, lalu menentukan penulisan berbingkai dengan memanggil TextIO.Write.withWindowedWrites.

Python

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

import argparse
from typing import List

import apache_beam as beam
from apache_beam.io.textio import WriteToText
from apache_beam.options.pipeline_options import PipelineOptions

from typing_extensions import Self


def write_to_cloud_storage(argv: List[str] = None) -> None:
    # Parse the pipeline options passed into the application.
    class MyOptions(PipelineOptions):
        @classmethod
        # Define a custom pipeline option that specfies the Cloud Storage bucket.
        def _add_argparse_args(cls: Self, parser: argparse.ArgumentParser) -> None:
            parser.add_argument("--output", required=True)

    wordsList = ["1", "2", "3", "4"]
    options = MyOptions()

    with beam.Pipeline(options=options.view_as(PipelineOptions)) as pipeline:
        (
            pipeline
            | "Create elements" >> beam.Create(wordsList)
            | "Write Files" >> WriteToText(options.output, file_name_suffix=".txt")
        )

Untuk jalur output, tentukan jalur Cloud Storage yang menyertakan nama bucket dan awalan nama file. Misalnya, jika Anda menentukan gs://my_bucket/output/file, konektor TextIO akan menulis ke bucket Cloud Storage bernama my_bucket, dan file output memiliki awalan output/file*.

Secara default, konektor TextIO mengelompokkan file output, menggunakan konvensi penamaan seperti ini: <file-prefix>-00000-of-00001. Secara opsional, Anda dapat menentukan akhiran nama file dan skema kompresi, seperti yang ditunjukkan dalam contoh.

Untuk memastikan penulisan idempoten, Dataflow menulis ke file sementara, lalu menyalin file sementara yang telah selesai ke file akhir. Untuk mengontrol tempat file sementara ini disimpan, gunakan metode withTempDirectory.

Langkah selanjutnya