Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

設定 Dataflow 管道選項

本頁面說明如何為 Dataflow 工作設定管道選項。這些管道選項會設定管道的執行方式、位置和使用的資源。

管道執行與 Apache Beam 程式執行是分開的。您編寫的 Apache Beam 程式會建構管道，以便延遲執行。也就是說，程式會產生一系列步驟，任何支援的 Apache Beam 執行器都能執行。相容的執行工具包括Google Cloud 上的 Dataflow 執行工具，以及直接在本機環境中執行管道的直接執行工具。

您可以在執行階段將參數傳遞至 Dataflow 工作。如要進一步瞭解如何在執行階段設定管道選項，請參閱「設定管道選項」。

注意：如果您要建立接受管道選項的傳統範本，則必須在管道程式碼中使用 ValueProvider 介面。詳情請參閱「在管道程式碼中使用執行階段參數」。

搭配使用管道選項和 Apache Beam SDK

您可以使用下列 SDK，為 Dataflow 工作設定管道選項：

Apache Beam SDK for Python
Java 適用的 Apache Beam SDK
Apache Beam SDK for Go

如要使用 SDK，請透過 Apache Beam SDK 類別 PipelineOptions 設定管道執行工具和其他執行參數。

指定管道選項的方法有兩種：

提供管道選項清單，以程式輔助方式設定管道選項。
執行管道程式碼時，直接在指令列設定管道選項。

以程式輔助方式設定管道選項

您可以建立及修改 PipelineOptions 物件，透過程式輔助方式設定管道選項。

Java

使用 PipelineOptionsFactory.fromArgs 方法建構 PipelineOptions 物件。

如需範例，請參閱本頁的「在 Dataflow 上啟動範例」一節。

Python

建立 PipelineOptions 物件。

如需範例，請參閱本頁的「在 Dataflow 上啟動範例」一節。

Go

Apache Beam SDK for Go 不支援使用 PipelineOptions 以程式輔助方式設定管道選項。請使用 Go 指令列引數。

如需範例，請參閱本頁的「在 Dataflow 上啟動範例」一節。

在指令列設定管道選項

您可以使用指令列引數設定管道選項。

Java

以下語法範例來自 Java 教學課程中的 WordCount 管道。

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"

更改下列內容：

PROJECT_ID：您的 Google Cloud 專案 ID
BUCKET_NAME：Cloud Storage bucket 的名稱
REGION：Dataflow 區域，us-central1

Python

以下語法範例來自 Python 教學課程中的 WordCount 管道。

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/

更改下列內容：

DATAFLOW_REGION：您要部署 Dataflow 工作的區域，例如 europe-west1

--region 旗標會覆寫中繼資料伺服器、本機用戶端或環境變數中設定的預設區域。
STORAGE_BUCKET：Cloud Storage bucket 名稱
PROJECT_ID： Google Cloud 專案 ID

Go

以下語法範例來自 Go 教學課程中的 WordCount 管道。

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/

更改下列內容：

BUCKET_NAME：Cloud Storage bucket 名稱
PROJECT_ID： Google Cloud 專案 ID
DATAFLOW_REGION：您要部署 Dataflow 工作的區域，例如：europe-west1。 --region 旗標會覆寫中繼資料伺服器、本機用戶端或環境變數中設定的預設區域。

設定實驗性管道選項

在 Java、Python 和 Go SDK 中，experiments管道選項可啟用實驗性或正式發布前的 Dataflow 功能。

透過程式輔助方式設定

如要透過程式設定 experiments 選項，請使用下列語法。

Java

在 PipelineOptions 物件中，使用下列語法加入 experiments 選項。以下範例使用實驗旗標，將開機磁碟大小設為 80 GB。

options.setExperiments("streaming_boot_disk_size_gb=80")

如需建立 PipelineOptions 物件的範例，請參閱本頁面的「在 Dataflow 上啟動範例」一節。

Python

在 PipelineOptions 物件中，使用下列語法加入 experiments 選項。以下範例使用實驗旗標，將開機磁碟大小設為 80 GB。

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

如需建立 PipelineOptions 物件的範例，請參閱本頁面的「在 Dataflow 上啟動範例」一節。

Go

Apache Beam SDK for Go 不支援使用 PipelineOptions 以程式輔助方式設定管道選項。請使用 Go 指令列引數。

透過指令列設定

如要在指令列設定 experiments 選項，請使用下列語法。

Java

以下範例使用實驗旗標，將開機磁碟大小設為 80 GB。

--experiments=streaming_boot_disk_size_gb=80

Python

以下範例使用實驗旗標，將開機磁碟大小設為 80 GB。

--experiments=streaming_boot_disk_size_gb=80

Go

以下範例使用實驗旗標，將開機磁碟大小設為 80 GB。

--experiments=streaming_boot_disk_size_gb=80

在範本中設定

如要在執行 Dataflow 範本時啟用實驗功能，請使用 --additional-experiments 旗標。

傳統範本

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

Flex 範本

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

存取管道選項物件

在 Apache Beam 程式中建立 Pipeline 物件時，請傳遞 PipelineOptions。Dataflow 服務執行管道時，會將 PipelineOptions 的副本傳送給每個 worker。

Java

使用 ProcessContext.getPipelineOptions 方法，在任何 ParDo 轉換的 DoFn 執行個體中存取 PipelineOptions。

Python

Apache Beam SDK for Python 不支援這項功能。

Go

使用 beam.PipelineOptions 存取管道選項。

在 Dataflow 啟動

使用 Dataflow 執行工具服務，在代管 Google Cloud 資源上執行工作。使用 Dataflow 執行管道時，系統會建立 Dataflow 工作，並使用 Google Cloud專案中的 Compute Engine 和 Cloud Storage 資源。如要瞭解 Dataflow 權限，請參閱「Dataflow 安全性與權限」。

在執行 pipeline 期間，Dataflow 工作使用 Cloud Storage 儲存暫存檔案。請關閉 Dataflow 暫存空間使用的 bucket 虛刪除功能，以免產生不必要的儲存費用。詳情請參閱「停用虛刪除功能」。

設定必要選項

如要使用 Dataflow 執行管道，請設定下列管道選項：

Java

project： Google Cloud 專案的 ID。
runner：執行管道的管道執行工具。如果是Google Cloud 執行作業，則必須是 DataflowRunner。
gcpTempLocation：Cloud Storage 路徑，Dataflow 會在此暫存大部分臨時檔案。指定的 bucket 必須已存在。

如未指定 gcpTempLocation，Dataflow 會使用 tempLocation 選項的值。如未指定上述任一選項，Dataflow 會建立新的 Cloud Storage bucket。

Python

project：您的 Google Cloud 專案 ID。
region：Dataflow 工作的區域。
runner：執行管道的管道執行工具。如果是Google Cloud 執行作業，則必須是 DataflowRunner。
temp_location：Cloud Storage 路徑，Dataflow 會將執行管道期間建立的暫時工作檔案暫存於此。

Go

project：您的 Google Cloud 專案 ID。
region：Dataflow 工作的區域。
runner：執行管道的管道執行工具。如果是Google Cloud 執行作業，則必須是 dataflow。
staging_location：Cloud Storage 路徑，Dataflow 會將執行管道期間建立的暫時工作檔案暫存於此。

以程式輔助方式設定管道選項

下列程式碼範例說明如何透過程式輔助方式設定執行工具和其他必要選項，建構管道，以便使用 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

Go 專用 Apache Beam SDK 使用 Go 指令列引數。使用 flag.Set() 設定旗標值。

// 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()

建構管道後，請指定所有管道讀取、轉換和寫入作業，然後執行管道。

透過指令列使用管道選項

以下範例說明如何使用在指令列中指定的管道選項。這個範例不會透過程式輔助方式設定管道選項。

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

使用 Python argparse 模組剖析指令列選項。

# 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

使用 Go flag 套件剖析指令列選項。您必須先剖析選項，才能呼叫 beam.Init()。在這個範例中，output 是指令列選項。

// 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()

建構管道後，請指定所有管道讀取、轉換和寫入作業，然後執行管道。

設定其他管道選項

除了直接使用 Apache Beam SDK 執行管道時設定選項外，您也可以在使用 gcloud CLI (例如執行範本或 YAML 管道) 或 Dataflow REST API 啟動工作時，設定管道和執行選項。

注意：如需管道選項和旗標的完整清單，請參閱「管道選項」。

下列分頁說明如何使用 gcloud CLI 和 REST API，設定常見的工作站、網路和執行選項。

gcloud

使用 gcloud dataflow jobs run、gcloud dataflow flex-template run 或 gcloud dataflow yaml run 等指令啟動工作或範本時，請使用指令列旗標指定執行和 worker 設定。

如要進一步瞭解如何設定 Flex 範本選項，請參閱「設定 Flex 範本的管道選項」。

如需各個 gcloud CLI 指令旗標的完整參考資料，請參閱 gcloud dataflow 參考文件。

API

如要使用 Dataflow REST API 啟動工作或範本，並指定執行和 worker 選項，請參閱下列參考頁面：

控管執行模式

當 Apache Beam 程式在 Dataflow 等服務上執行管道時，程式可以非同步執行管道，也可以封鎖直到管道完成。您可以按照下列指引變更這項行為。

Java

當 Apache Beam Java 程式在 Dataflow 等服務上執行管道時，通常會以非同步方式執行。如要執行管道並等待工作完成，請將 DataflowRunner 設為管道執行工具，並明確呼叫 pipeline.run().waitUntilFinish()。

使用 DataflowRunner 並在從 pipeline.run() 傳回的 PipelineResult 物件上呼叫 waitUntilFinish() 時，管道會在 Google Cloud 上執行，但本機程式碼會等待雲端工作完成並傳回最終 DataflowPipelineJob 物件。工作執行期間，Dataflow 服務會列印工作狀態更新和控制台訊息，同時等待。

Python

當 Apache Beam Python 程式在 Dataflow 等服務上執行管道時，通常會以非同步方式執行。如要封鎖至管道完成為止，請使用 PipelineResult 物件的 wait_until_finish() 方法，該物件會從執行器的 run() 方法傳回。

Go

Apache Beam Go 程式在 Dataflow 上執行管道時，預設會同步執行，並在管道完成前封鎖。如果不想封鎖，可以選擇以下兩種做法：

在 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.

使用 jobopts 套件中的 --async 指令列旗標。

如要查看執行詳細資料、監控進度及驗證工作完成狀態，請使用 Dataflow 監控介面或 Dataflow 指令列介面。

使用串流來源

Java

如果管道從無限制資料來源 (例如 Pub/Sub) 讀取資料，管道會自動以串流模式執行。

Python

如果管道使用無限制的資料來源 (例如 Pub/Sub)，則必須將 streaming 選項設為 true。

Go

如果管道從無限制資料來源 (例如 Pub/Sub) 讀取資料，管道會自動以串流模式執行。

串流工作預設會使用具有 2 個以上 vCPU 的 Compute Engine 機器類型。

在本機啟動

您可以選擇在本機執行管道，而非在代管雲端資源上執行。在本機執行管道，對於測試、偵錯或在小型資料集上執行管道，具有特定優勢。舉例來說，本機執行作業會移除對遠端 Dataflow 服務和相關聯 Google Cloud 專案的依附元件。

在本機執行時，您必須使用足夠小的資料集執行管道，才能放入本機記憶體。您可以使用 Create 轉換建立小型記憶體內資料集，或使用 Read 轉換處理小型本機或遠端檔案。本機執行作業通常能以更快速、更簡單的方式執行測試和偵錯，且外部依附元件較少，但受限於本機環境中的可用記憶體。

下列程式碼範例說明如何建構在本地環境中執行的管道。

Java

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

// 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')
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))

Go

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

beam.Init()

p := beam.NewPipeline()

將管道建構完成後，您就可以加以執行。

建立自訂管道選項

除了標準 PipelineOptions 之外，您也可以新增自訂選項。Apache Beam 的指令列也可以使用以相同格式指定的指令列引數，剖析自訂選項。

Java

如要新增自己的選項，請為每個選項定義具有 getter 和 setter 方法的介面，如以下範例所示：

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

Python

如要新增自己的選項，請使用 add_argument() 方法 (其行為與 Python 的標準 argparse 模組完全相同)，如以下範例所示：

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

如要新增自己的選項，請使用 Go 旗標套件，如以下範例所示：

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

您也可以指定說明 (當使用者傳遞 --help 做為指令列引數時顯示) 和預設值。

Java

請利用註解來設定說明和預設值，範例如下：

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

建議您向 PipelineOptionsFactory 註冊介面，然後在建立 PipelineOptions 物件時傳遞介面。向 PipelineOptionsFactory 註冊介面後，--help 就能找到自訂選項介面，並將其新增至 --help 指令的輸出內容。PipelineOptionsFactory 會驗證自訂選項是否與所有其他已註冊的選項相容。

以下程式碼範例說明如何向 PipelineOptionsFactory 註冊自訂選項介面：

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

現在，管道可以接受 --myCustomOption=value 做為指令列引數。

Python

請按照以下方式設定說明和預設值：

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

請按照以下方式設定說明和預設值：

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")
)

設定 Dataflow 管道選項 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

搭配使用管道選項和 Apache Beam SDK

以程式輔助方式設定管道選項

Java

Python

Go

在指令列設定管道選項

Java

Python

Go

設定實驗性管道選項

透過程式輔助方式設定

Java

Python

Go

透過指令列設定

Java

Python

Go

在範本中設定

傳統範本

Flex 範本

存取管道選項物件

Java

Python

Go

在 Dataflow 啟動

設定必要選項

Java

Python

Go

以程式輔助方式設定管道選項

Java

Python

Go

透過指令列使用管道選項

Java

Python

Go

設定其他管道選項

gcloud

API

控管執行模式

Java

Python

Go

使用串流來源

Java

Python

Go

在本機啟動

Java

Python

Go

建立自訂管道選項

Java

Python

Go

Java

Python

Go

設定 Dataflow 管道選項