Dataflow에서 Cloud Storage에 쓰기

이 문서에서는 Apache Beam TextIO I/O 커넥터를 사용하여 Dataflow의 텍스트 데이터를 Cloud Storage로 쓰는 방법을 설명합니다.

Google Cloud 라이브러리 종속 항목 포함

Cloud Storage에 TextIO 커넥터를 사용하려면 다음 종속 항목을 포함합니다. 이 라이브러리는 "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"

자세한 내용은 Apache Beam SDK 설치를 참조하세요.

Dataflow의 Apache Beam I/O 커넥터에서 gRPC 사용 설정

Dataflow의 Apache Beam I/O 커넥터를 통해 gRPC를 사용하여 Cloud Storage에 연결할 수 있습니다. gRPC는 Google에서 개발한 고성능 오픈소스 원격 프로시저 호출 (RPC) 프레임워크로, Cloud Storage와 상호작용하는 데 사용할 수 있습니다.

Dataflow 작업의 Cloud Storage 쓰기 요청 속도를 높이려면 Dataflow에서 Apache Beam I/O 커넥터를 사용 설정하여 gRPC를 사용하면 됩니다.

Dataflow에서 Apache Beam I/O 커넥터를 구성하여 Cloud Monitoring에서 gRPC 관련 측정항목을 생성할 수 있습니다. gRPC 관련 측정항목을 사용하면 다음을 수행할 수 있습니다.

• Cloud Storage에 대한 gRPC 요청의 성능을 모니터링하고 최적화합니다.
• 문제 해결 및 디버깅
• 애플리케이션의 사용 및 동작에 대한 유용한 정보를 파악합니다.

동시 로드

동시 로드는 주로 샤드 수에 따라 결정됩니다. 기본적으로 런너는 이 값을 자동으로 설정합니다. 대부분의 파이프라인에서는 기본 동작을 사용하는 것이 좋습니다. 이 문서에서 권장사항을 참고하세요.

성능

다음 표는 Cloud Storage에 쓰기의 성능 측정항목을 보여줍니다. 워크로드는 자바용 Apache Beam SDK 2.49.0을 사용해 하나의 e2-standard2 작업자에서 실행되었습니다. Runner v2를 사용하지 않았습니다.

이러한 측정항목은 단순 배치 파이프라인을 기반으로 합니다. 이러한 측정항목은 I/O 커넥터 사이의 성능 비교를 위해 사용되며 반드시 실제 파이프라인을 나타내지는 않습니다. Dataflow 파이프라인 성능은 복잡하며 VM 유형, 처리 중인 데이터, 외부 소스 및 싱크의 성능, 사용자 코드와 상관관계가 있습니다. 측정항목은 Java SDK 실행을 기반으로 하며 다른 언어 SDK의 성능 특성을 나타내지 않습니다. 자세한 내용은 Beam IO 성능을 참조하세요.

권장사항

• 일반적으로 특정 샤드 수를 설정하지 않는 것이 좋습니다. 이렇게 하면 러너가 규모에 적절한 값을 선택할 수 있습니다. 샤드 수를 조정하는 경우 샤드당 100MB~1GB를 쓰는 것이 좋습니다. 그러나 최적의 값은 워크로드에 따라 달라질 수 있습니다.

• Cloud Storage는 초당 매우 많은 수의 요청으로 확장할 수 있습니다. 하지만 파이프라인의 쓰기 볼륨이 급증하는 경우 단일 Cloud Storage 버킷에 일시적으로 과부하가 발생하지 않도록 여러 버킷에 쓰는 것이 좋습니다.

• 일반적으로 각 쓰기 크기가 더 클수록 (1KB 이상) Cloud Storage에 쓰는 것이 더 효율적입니다. 작은 레코드를 다수의 파일에 쓰면 바이트당 성능이 저하될 수 있습니다.

• 파일 이름을 생성할 때는 부하를 분산하기 위해 순서가 아닌 파일 이름을 사용하는 것이 좋습니다. 자세한 내용은 전체 키 범위에 걸쳐 부하를 균일하게 분배하는 이름 지정 규칙 사용을 참고하세요.

• 파일 이름을 지정할 때는 @ 기호 ('@') 뒤에 숫자나 별표 ('*')를 사용하지 마세요. 자세한 내용은 '@*' 및 '@N'은 예약된 샤딩 사양입니다를 참고하세요.

예: Cloud Storage에 텍스트 파일 쓰기

다음 예시에서는 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")
        )

출력 경로의 경우 버킷 이름과 파일 이름 접두사가 포함된 Cloud Storage 경로를 지정합니다. 예를 들어 gs://my_bucket/output/file을 지정하면 TextIO 커넥터는 my_bucket이라는 Cloud Storage 버킷을 대상으로 쓰고 출력 파일에는 output/file* 접두사가 포함됩니다.

기본적으로 TextIO 커넥터는 <file-prefix>-00000-of-00001과 같은 이름 지정 규칙을 사용하여 출력 파일을 샤딩합니다. 원하는 경우 예시와 같이 파일 이름 확장자와 압축 스키마를 지정할 수 있습니다.

일관된 쓰기를 보장하기 위해 Dataflow는 임시 파일에 쓰고 완료된 임시 파일을 최종 파일에 복사합니다. 이러한 임시 파일이 저장되는 위치를 제어하려면 withTempDirectory 메서드를 사용하세요.

다음 단계

• TextIO API 문서 참조
• Google 제공 템플릿 목록 참조