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 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 mengetahui informasi selengkapnya, lihat Menginstal Apache Beam SDK.

Keparalelan

Paralelisme ditentukan terutama oleh jumlah shard. Secara default, runner secara 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 penulisan ke Cloud Storage. Workload dijalankan pada satu pekerja e2-standard2, menggunakan Apache Beam SDK 2.49.0 untuk Java. Dia tidak menggunakan Runner v2.

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

Metrik ini didasarkan pada pipeline batch sederhana. Keduanya ditujukan untuk membandingkan performa antara konektor I/O, dan tidak selalu merepresentasikan 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. Ini memungkinkan pelari untuk memilih nilai yang sesuai untuk skala Anda. Jika Anda menyesuaikan jumlah shard, sebaiknya tulis antara 100 MB hingga 1 GB per shard. Namun, nilai optimal mungkin bergantung pada beban kerja.

  • Cloud Storage dapat menskalakan hingga jumlah permintaan per detik yang sangat besar. Namun, jika pipeline Anda mengalami lonjakan besar dalam volume tulis, pertimbangkan untuk menulis ke beberapa bucket agar tidak kelebihan beban untuk sementara waktu pada satu bucket Cloud Storage.

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

  • Saat membuat nama file, pertimbangkan untuk menggunakan nama file yang tidak berurutan agar dapat 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 @ ('@') diikuti 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 mengautentikasi 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 berjendela dengan memanggil TextIO.Write.withWindowedWrites.

Python

Untuk mengautentikasi 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) 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 melakukan sharding file output, menggunakan konvensi penamaan seperti ini: <file-prefix>-00000-of-00001. Jika ingin, Anda dapat menentukan akhiran nama file dan skema kompresi, seperti yang ditunjukkan dalam contoh.

Untuk memastikan operasi tulis idempoten, Dataflow menulis ke file sementara, lalu menyalin file sementara yang sudah selesai ke file akhir.

Langkah selanjutnya