Menetapkan opsi pipeline Dataflow

Halaman ini menjelaskan cara menetapkan opsi pipeline untuk tugas Dataflow Anda. Opsi pipeline ini mengonfigurasi cara dan tempat pipeline Anda berjalan serta resource yang digunakannya.

Eksekusi pipeline terpisah dari eksekusi program Apache Beam Anda. Program Apache Beam yang telah Anda tulis membuat pipeline untuk eksekusi yang ditangguhkan. Artinya, program ini menghasilkan serangkaian langkah yang dapat dijalankan oleh runner Apache Beam yang didukung. Runner yang kompatibel mencakup runner Dataflow di Google Cloud dan runner langsung yang mengeksekusi pipeline secara langsung di lingkungan lokal.

Anda dapat meneruskan parameter ke tugas Dataflow saat runtime. Untuk informasi tambahan tentang cara menetapkan opsi pipeline saat runtime, lihat Mengonfigurasi opsi pipeline.

Catatan: Jika Anda membuat template klasik yang menerima opsi pipeline, Anda harus menggunakan antarmuka ValueProvider dalam kode pipeline. Untuk mengetahui informasi selengkapnya, lihat Menggunakan parameter runtime dalam kode pipeline Anda.

Menggunakan opsi pipeline dengan Apache Beam SDK

Anda dapat menggunakan SDK berikut untuk menetapkan opsi pipeline untuk tugas Dataflow:

• Apache Beam SDK untuk Python
• Apache Beam SDK untuk Java
• Apache Beam SDK untuk Go

Untuk menggunakan SDK, Anda menetapkan runner pipeline dan parameter eksekusi lainnya menggunakan class Apache Beam SDK PipelineOptions.

Ada dua metode untuk menentukan opsi pipeline:

• Tetapkan opsi pipeline secara terprogram dengan menyediakan daftar opsi pipeline.
• Tetapkan opsi pipeline langsung di command line saat Anda menjalankan kode pipeline.

Menetapkan opsi pipeline secara terprogram

Anda dapat menetapkan opsi pipeline secara terprogram dengan membuat dan mengubah objek PipelineOptions.

Java

Buat objek PipelineOptions menggunakan metode PipelineOptionsFactory.fromArgs.

Sebagai contoh, lihat bagian Contoh peluncuran di Dataflow di halaman ini.

Python

Buat objek PipelineOptions.

Sebagai contoh, lihat bagian Contoh peluncuran di Dataflow di halaman ini.

Go

Menetapkan opsi pipeline secara terprogram menggunakan PipelineOptions tidak didukung di Apache Beam SDK untuk Go. Gunakan argumen command line Go.

Sebagai contoh, lihat bagian Contoh peluncuran di Dataflow di halaman ini.

Menetapkan opsi pipeline di command line

Anda dapat menetapkan opsi pipeline menggunakan argumen command line.

Java

Contoh sintaksis berikut berasal dari pipeline WordCount dalam tutorial Java.

mvn -Pdataflow-runner compile exec:java \
  -Dexec.mainClass=org.apache.beam.examples.WordCount \
  -Dexec.args="--project=PROJECT_ID \
  --gcpTempLocation=gs://BUCKET_NAME/temp/ \
  --output=gs://BUCKET_NAME/output \
  --runner=DataflowRunner \
  --region=REGION"

Ganti kode berikut:

• PROJECT_ID: Project ID Google Cloud Anda
• BUCKET_NAME: nama bucket Cloud Storage Anda
• REGION: region Dataflow, us-central1

Python

Contoh sintaksis berikut berasal dari pipeline WordCount dalam tutorial Python.

python -m apache_beam.examples.wordcount \
  --region DATAFLOW_REGION \
  --input gs://dataflow-samples/shakespeare/kinglear.txt \
  --output gs://STORAGE_BUCKET/results/outputs \
  --runner DataflowRunner \
  --project PROJECT_ID \
  --temp_location gs://STORAGE_BUCKET/tmp/

Ganti kode berikut:

• DATAFLOW_REGION: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, europe-west1

  Flag --region menggantikan region default yang ditetapkan di server metadata, klien lokal, atau variabel lingkungan.

• STORAGE_BUCKET: nama bucket Cloud Storage

• PROJECT_ID: Project ID Google Cloud

Go

Contoh sintaksis berikut berasal dari pipeline WordCount dalam tutorial Go.

go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
   --output gs://BUCKET_NAME/results/outputs \
   --runner dataflow \
   --project PROJECT_ID \
   --region DATAFLOW_REGION \
   --staging_location gs://BUCKET_NAME/binaries/

Ganti kode berikut:

• BUCKET_NAME: nama bucket Cloud Storage

• PROJECT_ID: Project ID Google Cloud

• DATAFLOW_REGION: Region tempat Anda ingin men-deploy tugas Dataflow. Misalnya, europe-west1. Flag --region menggantikan region default yang ditetapkan di server metadata, klien lokal, atau variabel lingkungan.

Menetapkan opsi pipeline eksperimental

Di SDK Java, Python, dan Go, opsi pipeline experiments mengaktifkan fitur Dataflow eksperimental atau pra-GA.

Menetapkan secara terprogram

Untuk menetapkan opsi experiments secara terprogram, gunakan sintaksis berikut.

Java

Dalam objek PipelineOptions Anda, sertakan opsi experiments menggunakan sintaksis berikut. Contoh ini menetapkan ukuran boot disk ke 80 GB dengan flag eksperimen.

options.setExperiments("streaming_boot_disk_size_gb=80")

Untuk contoh yang menunjukkan cara membuat objek PipelineOptions, lihat bagian Contoh peluncuran di Dataflow di halaman ini.

Python

Dalam objek PipelineOptions Anda, sertakan opsi experiments menggunakan sintaksis berikut. Contoh ini menetapkan ukuran boot disk ke 80 GB dengan flag eksperimen.

beam_options = PipelineOptions(
  beam_args,
  experiments=['streaming_boot_disk_size_gb=80'])

Untuk contoh yang menunjukkan cara membuat objek PipelineOptions, lihat bagian Contoh peluncuran di Dataflow di halaman ini.

Go

Menetapkan opsi pipeline secara terprogram menggunakan PipelineOptions tidak didukung di Apache Beam SDK untuk Go. Gunakan argumen command line Go.

Ditetapkan di command line

Untuk menetapkan opsi experiments di command line, gunakan sintaksis berikut.

Java

Contoh ini menetapkan ukuran boot disk ke 80 GB dengan flag eksperimen.

--experiments=streaming_boot_disk_size_gb=80

Python

Contoh ini menetapkan ukuran boot disk ke 80 GB dengan flag eksperimen.

--experiments=streaming_boot_disk_size_gb=80

Go

Contoh ini menetapkan ukuran boot disk ke 80 GB dengan flag eksperimen.

--experiments=streaming_boot_disk_size_gb=80

Ditetapkan dalam template

Untuk mengaktifkan fitur eksperimental saat menjalankan template Dataflow, gunakan flag --additional-experiments.

Template klasik

gcloud dataflow jobs run JOB_NAME --additional-experiments=EXPERIMENT[,...]

Template Flex

gcloud dataflow flex-template run JOB_NAME --additional-experiments=EXPERIMENT[,...]

Mengakses objek opsi pipeline

Saat Anda membuat objek Pipeline di program Apache Beam, teruskan PipelineOptions. Saat layanan Dataflow menjalankan pipeline Anda, layanan ini akan mengirimkan salinan PipelineOptions ke setiap worker.

Java

Akses PipelineOptions di dalam instance DoFn transformasi ParDo apa pun menggunakan metode ProcessContext.getPipelineOptions.

Python

Fitur ini tidak didukung di Apache Beam SDK untuk Python.

Go

Akses opsi pipeline menggunakan beam.PipelineOptions.

Meluncurkan di Dataflow

Jalankan tugas Anda di resource Google Cloud terkelola menggunakan layanan runner Dataflow. Menjalankan pipeline dengan Dataflow akan membuat tugas Dataflow, yang menggunakan resource Compute Engine dan Cloud Storage di project Google CloudAnda. Untuk mengetahui informasi tentang izin Dataflow, lihat Keamanan dan izin Dataflow.

Tugas Dataflow menggunakan Cloud Storage untuk menyimpan file sementara selama eksekusi pipeline. Agar tidak dikenai biaya penyimpanan yang tidak perlu, nonaktifkan fitur penghapusan sementara pada bucket yang digunakan tugas Dataflow Anda untuk penyimpanan sementara. Untuk mengetahui informasi selengkapnya, lihat Menonaktifkan penghapusan sementara.

Catatan: Mengetik Ctrl+C dari command line tidak akan membatalkan tugas Anda. Layanan Dataflow masih menjalankan tugas di Google Cloud. Untuk membatalkan tugas, gunakan Antarmuka Pemantauan Dataflow atau antarmuka command line Dataflow.

Menetapkan opsi yang diperlukan

Untuk menjalankan pipeline menggunakan Dataflow, tetapkan opsi pipeline berikut:

Java

• project: ID project Google Cloud Anda.
• runner: runner pipeline yang mengeksekusi pipeline Anda. Untuk eksekusi Google Cloud , nilai ini harus berupa DataflowRunner.

• gcpTempLocation: jalur Cloud Storage untuk Dataflow guna melakukan staging sebagian besar file sementara. Bucket yang ditentukan harus sudah ada.

  Jika Anda tidak menentukan gcpTempLocation, Dataflow akan menggunakan nilai opsi tempLocation. Jika Anda tidak menentukan salah satu opsi ini, Dataflow akan membuat bucket Cloud Storage baru.

Python

• project: Google Cloud project ID Anda.
• region: region untuk tugas Dataflow Anda.
• runner: runner pipeline yang mengeksekusi pipeline Anda. Untuk eksekusi Google Cloud , nilai ini harus berupa DataflowRunner.
• temp_location: jalur Cloud Storage untuk Dataflow guna melakukan staging file tugas sementara yang dibuat selama eksekusi pipeline.

Go

• project: Google Cloud project ID Anda.
• region: region untuk tugas Dataflow Anda.
• runner: runner pipeline yang mengeksekusi pipeline Anda. Untuk eksekusi Google Cloud , nilai ini harus berupa dataflow.
• staging_location: jalur Cloud Storage untuk Dataflow guna men-stage file tugas sementara yang dibuat selama eksekusi pipeline.

Menetapkan opsi pipeline secara terprogram

Kode contoh berikut menunjukkan cara membuat pipeline dengan menetapkan runner dan opsi lain yang diperlukan secara terprogram untuk menjalankan pipeline menggunakan Dataflow.

Java

// Create and set your PipelineOptions.
DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);

// For cloud execution, set the Google Cloud project, staging location,
// and set DataflowRunner.
options.setProject("my-project-id");
options.setStagingLocation("gs://my-bucket/binaries");
options.setRunner(DataflowRunner.class);

// Create the Pipeline with the specified options.
Pipeline p = Pipeline.create(options);

Python

import argparse

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

parser = argparse.ArgumentParser()
# parser.add_argument('--my-arg', help='description')
args, beam_args = parser.parse_known_args()

# Create and set your PipelineOptions.
# For Cloud execution, specify DataflowRunner and set the Cloud Platform
# project, job name, temporary files location, and region.
# For more information about regions, check:
# https://cloud.google.com/dataflow/docs/concepts/regional-endpoints
beam_options = PipelineOptions(
    beam_args,
    runner='DataflowRunner',
    project='my-project-id',
    job_name='unique-job-name',
    temp_location='gs://my-bucket/temp',
    region='us-central1')
# Note: Repeatable options like dataflow_service_options or experiments must
# be specified as a list of string(s).
# e.g. dataflow_service_options=['enable_prime']

# Create the Pipeline with the specified options.
with beam.Pipeline(options=beam_options) as pipeline:
  pass  # build your pipeline here.

Go

Apache Beam SDK untuk Go menggunakan argumen command line Go. Gunakan flag.Set() untuk menetapkan nilai flag.

// Use the Go flag package to parse custom options.
flag.Parse()

// Set the required options programmatically.
// For Cloud execution, specify the Dataflow runner, Google Cloud
// project ID, region, and staging location.
// For more information about regions, see
// https://cloud.google.com/dataflow/docs/concepts/regional-endpoints
flag.Set("runner", "dataflow")
flag.Set("project", "my-project-id")
flag.Set("region", "us-central1")
flag.Set("staging_location", "gs://my-bucket/binaries")

beam.Init()

// Create the Pipeline.
p := beam.NewPipeline()
s := p.Root()

Setelah membuat pipeline, tentukan semua pembacaan, transformasi, dan penulisan pipeline, lalu jalankan pipeline.

Menggunakan opsi pipeline dari command line

Contoh berikut menunjukkan cara menggunakan opsi pipeline yang ditentukan di command line. Contoh ini tidak menetapkan opsi pipeline secara terprogram.

Java

// Set your PipelineOptions to the specified command-line options
MyOptions options = PipelineOptionsFactory.fromArgs(args).withValidation();

// Create the Pipeline with the specified options.
Pipeline p = Pipeline.create(options);

Python

Gunakan modul argparse Python untuk mengurai opsi command line.

# Use Python argparse module to parse custom arguments
import argparse

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

# For more details on how to use argparse, take a look at:
#   https://docs.python.org/3/library/argparse.html
parser = argparse.ArgumentParser()
parser.add_argument(
    '--input-file',
    default='gs://dataflow-samples/shakespeare/kinglear.txt',
    help='The file path for the input text to process.')
parser.add_argument(
    '--output-path', required=True, help='The path prefix for output files.')
args, beam_args = parser.parse_known_args()

# Create the Pipeline with remaining arguments.
beam_options = PipelineOptions(beam_args)
with beam.Pipeline(options=beam_options) as pipeline:
  lines = (
      pipeline
      | 'Read files' >> beam.io.ReadFromText(args.input_file)
      | 'Write files' >> beam.io.WriteToText(args.output_path))

Go

Gunakan paket Go flag untuk mengurai opsi command line. Anda harus mengurai opsi sebelum memanggil beam.Init(). Dalam contoh ini, output adalah opsi command line.

// Define configuration options
var (
  output = flag.String("output", "", "Output file (required).")
)

// Parse options before beam.Init()
flag.Parse()

beam.Init()

// Input validation must be done after beam.Init()
if *output == "" {
  log.Fatal("No output provided!")
}

p := beam.NewPipeline()

Setelah membuat pipeline, tentukan semua pembacaan, transformasi, dan penulisan pipeline, lalu jalankan pipeline.

Menetapkan opsi pipeline tambahan

Selain menetapkan opsi saat menjalankan pipeline secara langsung menggunakan Apache Beam SDK, Anda dapat mengonfigurasi opsi pipeline dan eksekusi saat meluncurkan tugas menggunakan gcloud CLI (seperti saat menjalankan template atau pipeline YAML) atau Dataflow REST API.

Catatan: Untuk mengetahui daftar lengkap opsi dan flag pipeline, lihat Opsi pipeline.

Tab berikut menunjukkan cara mengonfigurasi opsi worker, jaringan, dan eksekusi umum menggunakan gcloud CLI dan REST API.

gcloud

Saat meluncurkan tugas atau template menggunakan perintah seperti gcloud dataflow jobs run, gcloud dataflow flex-template run, atau gcloud dataflow yaml run, gunakan flag command line untuk menentukan setelan eksekusi dan worker.

Untuk mengetahui informasi selengkapnya tentang cara menetapkan opsi Template Flex, lihat Menetapkan opsi pipeline untuk Template Flex.

Untuk referensi lengkap flag untuk setiap perintah gcloud CLI, lihat dokumentasi referensi gcloud dataflow.

API

Untuk menentukan opsi eksekusi dan worker saat meluncurkan tugas atau template menggunakan Dataflow REST API, lihat halaman referensi berikut:

• Metode: projects.locations.jobs.create
• Method: projects.locations.templates.create
• Metode: projects.locations.flexTemplates.launch

Mengontrol mode eksekusi

Saat program Apache Beam menjalankan pipeline di layanan seperti Dataflow, program dapat menjalankan pipeline secara asinkron, atau dapat memblokir hingga pipeline selesai. Anda dapat mengubah perilaku ini dengan menggunakan panduan berikut.

Java

Saat program Java Apache Beam menjalankan pipeline di layanan seperti Dataflow, pipeline tersebut biasanya dijalankan secara asinkron. Untuk menjalankan pipeline dan menunggu hingga tugas selesai, tetapkan DataflowRunner sebagai runner pipeline dan panggil pipeline.run().waitUntilFinish() secara eksplisit.

Saat Anda menggunakan DataflowRunner dan memanggil waitUntilFinish() pada objek PipelineResult yang ditampilkan dari pipeline.run(), pipeline akan dieksekusi di Google Cloud , tetapi kode lokal akan menunggu tugas cloud selesai dan menampilkan objek DataflowPipelineJob akhir. Saat tugas berjalan, layanan Dataflow akan mencetak update status tugas dan pesan konsol saat menunggu.

Python

Saat program Python Apache Beam menjalankan pipeline di layanan seperti Dataflow, program tersebut biasanya dijalankan secara asinkron. Untuk memblokir hingga pipeline selesai, gunakan metode wait_until_finish() dari objek PipelineResult, yang ditampilkan dari metode run() runner.

Go

Saat program Apache Beam Go menjalankan pipeline di Dataflow, program tersebut bersifat sinkron secara default dan memblokir hingga pipeline selesai. Jika Anda tidak ingin memblokir, ada dua opsi:

1. Mulai tugas dalam rutinitas Go.

   go func() {
     pr, err := beamx.Run(ctx, p)
     if err != nil {
       // Handle the error.
     }
     // Send beam.PipelineResult into a channel.
     results <- pr
   }()
   // Do other operations while the pipeline runs.
   

2. Gunakan flag command line --async, yang ada dalam paket jobopts.

Untuk melihat detail eksekusi, memantau progres, dan memverifikasi status penyelesaian tugas, gunakan antarmuka pemantauan Dataflow atau antarmuka command line Dataflow.

Menggunakan sumber streaming

Java

Jika pipeline Anda membaca dari sumber data tak terbatas, seperti Pub/Sub, pipeline akan otomatis dieksekusi dalam mode streaming.

Python

Jika pipeline Anda menggunakan sumber data yang tidak terbatas, seperti Pub/Sub, Anda harus menetapkan opsi streaming ke true.

Go

Jika pipeline Anda membaca dari sumber data tak terbatas, seperti Pub/Sub, pipeline akan otomatis dieksekusi dalam mode streaming.

Tugas streaming menggunakan jenis mesin Compute Engine dengan 2 vCPU atau lebih secara default.

Catatan: Jika pipeline Anda menggunakan sink dan sumber data unbounded, Anda harus memilih strategi Windowing untuk PCollection unbounded sebelum menggunakan agregasi apa pun seperti GroupByKey.

Meluncurkan secara lokal

Alih-alih menjalankan pipeline pada resource cloud terkelola, Anda dapat memilih untuk mengeksekusi pipeline secara lokal. Eksekusi lokal memiliki keunggulan tertentu untuk menguji, men-debug, atau menjalankan pipeline Anda pada set data kecil. Misalnya, eksekusi lokal menghapus dependensi pada layanan Dataflow jarak jauh dan project Google Cloud terkait.

Saat menggunakan eksekusi lokal, Anda harus menjalankan pipeline dengan set data yang cukup kecil agar muat dalam memori lokal. Anda dapat membuat set data dalam memori kecil menggunakan transformasi Create, atau Anda dapat menggunakan transformasi Read untuk menangani file lokal atau jarak jauh yang kecil. Eksekusi lokal biasanya memberikan cara yang lebih cepat dan mudah untuk melakukan pengujian dan proses debug dengan lebih sedikit dependensi eksternal, tetapi dibatasi oleh memori yang tersedia di lingkungan lokal Anda.

Kode contoh berikut menunjukkan cara membuat pipeline yang dijalankan di lingkungan lokal Anda.

Java

// Create and set our Pipeline Options.
PipelineOptions options = PipelineOptionsFactory.create();

// Create the Pipeline with the specified options.
Pipeline p = Pipeline.create(options);

Catatan: Untuk mode lokal, Anda tidak perlu menetapkan runner karena DirectRunner sudah menjadi default. Namun, Anda harus menyertakan DirectRunner secara eksplisit sebagai dependensi atau menambahkannya ke classpath.

Python

import argparse

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

parser = argparse.ArgumentParser()
# parser.add_argument('--my-arg')
args, beam_args = parser.parse_known_args()

# Create and set your Pipeline Options.
beam_options = PipelineOptions(beam_args)
args = beam_options.view_as(MyOptions)

with beam.Pipeline(options=beam_options) as pipeline:
  lines = (
      pipeline
      | beam.io.ReadFromText(args.input)
      | beam.io.WriteToText(args.output))

Catatan: Untuk mode lokal, Anda tidak perlu menetapkan runner karena DirectRunner sudah menjadi default.

Go

// Parse options before beam.Init()
flag.Parse()

beam.Init()

p := beam.NewPipeline()

Catatan: Untuk mode lokal, Anda tidak perlu menetapkan runner karena direct sudah menjadi default.

Setelah Anda membuat pipeline, jalankan pipeline tersebut.

Membuat opsi pipeline kustom

Anda dapat menambahkan opsi kustom Anda sendiri selain PipelineOptions standar. Command line Apache Beam juga dapat mengurai opsi kustom menggunakan argumen command line yang ditentukan dalam format yang sama.

Java

Untuk menambahkan opsi Anda sendiri, tentukan antarmuka dengan metode pengambil dan penyetel untuk setiap opsi, seperti dalam contoh berikut:

public interface MyOptions extends PipelineOptions {
  String getMyCustomOption();
  void setMyCustomOption(String myCustomOption);
}

Python

Untuk menambahkan opsi Anda sendiri, gunakan metode add_argument() (yang berperilaku persis seperti modul argparse standar Python), seperti dalam contoh berikut:

from apache_beam.options.pipeline_options import PipelineOptions

class MyOptions(PipelineOptions):
  @classmethod
  def _add_argparse_args(cls, parser):
    parser.add_argument('--input')
    parser.add_argument('--output')

Go

Untuk menambahkan opsi Anda sendiri, gunakan paket flag Go seperti yang ditunjukkan dalam contoh berikut:

var (
  input  = flag.String("input", "", "")
  output = flag.String("output", "", "")
)

Anda juga dapat menentukan deskripsi, yang muncul saat pengguna meneruskan --help sebagai argumen command line, dan nilai default.

Java

Anda menetapkan deskripsi dan nilai default menggunakan anotasi, seperti berikut:

public interface MyOptions extends PipelineOptions {
  @Description("My custom command line argument.")
  @Default.String("DEFAULT")
  String getMyCustomOption();
  void setMyCustomOption(String myCustomOption);
}

Sebaiknya daftarkan antarmuka Anda dengan PipelineOptionsFactory, lalu teruskan antarmuka saat membuat objek PipelineOptions. Saat Anda mendaftarkan antarmuka dengan PipelineOptionsFactory, --help dapat menemukan antarmuka opsi kustom Anda dan menambahkannya ke output perintah --help. PipelineOptionsFactory memvalidasi bahwa opsi kustom Anda kompatibel dengan semua opsi terdaftar lainnya.

Kode contoh berikut menunjukkan cara mendaftarkan antarmuka opsi kustom Anda dengan PipelineOptionsFactory:

PipelineOptionsFactory.register(MyOptions.class);
MyOptions options = PipelineOptionsFactory.fromArgs(args)
                                          .withValidation()
                                          .as(MyOptions.class);

Sekarang pipeline Anda dapat menerima --myCustomOption=value sebagai argumen command line.

Python

Anda menetapkan deskripsi dan nilai default sebagai berikut:

from apache_beam.options.pipeline_options import PipelineOptions

class MyOptions(PipelineOptions):
  @classmethod
  def _add_argparse_args(cls, parser):
    parser.add_argument(
        '--input',
        default='gs://dataflow-samples/shakespeare/kinglear.txt',
        help='The file path for the input text to process.')
    parser.add_argument(
        '--output', required=True, help='The path prefix for output files.')

Go

Anda menetapkan deskripsi dan nilai default sebagai berikut:

var (
  input  = flag.String("input", "gs://MY_STORAGE_BUCKET/input", "Input for the pipeline")
  output = flag.String("output", "gs://MY_STORAGE_BUCKET/output", "Output for the pipeline")
)