Menulis dari Dataflow ke Cloud Storage

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

Menyertakan dependensi library Google Cloud

Untuk menggunakan konektor TextIO 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 penulisan 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

Paralelisme ditentukan terutama oleh jumlah shard. Secara default, pelari 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.

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

• Secara umum, hindari menyetel jumlah shard tertentu. Hal ini memungkinkan pelari memilih nilai yang sesuai untuk skala Anda. Untuk mengaktifkan pemartisian otomatis, panggil .withAutoSharding(), bukan .⇉withNumShards⇇(0). Jika Anda menyesuaikan jumlah shard, sebaiknya tulis antara 100 MB dan 1 GB per shard. Namun, nilai optimal mungkin bergantung pada beban kerja.

• Cloud Storage dapat diskalakan untuk menangani permintaan dalam jumlah yang sangat besar per detik. Namun, jika pipeline Anda mengalami lonjakan volume penulisan yang besar, pertimbangkan untuk menulis ke beberapa bucket, untuk menghindari kelebihan muatan sementara pada satu bucket Cloud Storage.

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

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

• Saat memberi nama file, jangan gunakan tanda @ ('@') yang diikuti dengan angka atau tanda bintang ('*'). Untuk mengetahui 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:

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

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 mencakup 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 memecah 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 penyimpanan file sementara ini, gunakan metode withTempDirectory.

Langkah berikutnya

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