הגדרת אפשרויות של צינור עיבוד נתונים ב-Dataflow

בדף הזה מוסבר איך מגדירים אפשרויות של צינור עיבוד נתונים למשימות ב-Dataflow. האפשרויות האלה קובעות איך ואיפה הצינור יפעל ובאילו משאבים הוא ישתמש.

הפעלת צינור עיבוד הנתונים נפרדת מהפעלת תוכנית Apache Beam. תוכנית Apache Beam שכתבתם יוצרת צינור עיבוד נתונים להפעלה מושהית. כלומר, התוכנית יוצרת סדרה של שלבים שכל מריץ נתמך של Apache Beam יכול לבצע. בין ה-runners התואמים: ה-runner של Dataflow ב-Google Cloud וה-runner הישיר שמפעיל את צינור עיבוד הנתונים ישירות בסביבה מקומית.

אתם יכולים להעביר פרמטרים למשימת Dataflow בזמן הריצה. למידע נוסף על הגדרת אפשרויות צינור עיבוד נתונים בזמן הריצה, ראו הגדרת אפשרויות צינור עיבוד נתונים.

שימוש באפשרויות של צינורות עיבוד נתונים עם ערכות SDK של Apache Beam

אתם יכולים להשתמש בערכות ה-SDK הבאות כדי להגדיר אפשרויות לצינור עיבוד הנתונים של משימות Dataflow:

  • Apache Beam SDK ל-Python
  • Apache Beam SDK ל-Java
  • Apache Beam SDK ל-Go

כדי להשתמש בערכות ה-SDK, מגדירים את מריץ הצינור ופרמטרים אחרים של הפעלה באמצעות המחלקה PipelineOptions של Apache Beam SDK.

יש שתי שיטות להגדרת אפשרויות צינור עיבוד נתונים:

  • הגדרת אפשרויות של צינור עיבוד נתונים באופן פרוגרמטי על ידי ציון רשימה של אפשרויות לצינור עיבוד הנתונים.
  • מגדירים את האפשרויות של צינור עיבוד הנתונים ישירות בשורת הפקודה כשמריצים את הקוד של צינור עיבוד הנתונים.

הגדרת אפשרויות צינור עיבוד נתונים באופן פרוגרמטי

כדי להגדיר אפשרויות של צינור עיבוד נתונים באופן פרוגרמטי, אפשר ליצור ולשנות אובייקט PipelineOptions.

Java

יוצרים אובייקט PipelineOptions באמצעות ה-method‏ PipelineOptionsFactory.fromArgs.

לדוגמה, ראו את הקטע דוגמה להפעלה ב-Dataflow בדף הזה.

Python

יוצרים אובייקט PipelineOptions.

לדוגמה, ראו את הקטע דוגמה להפעלה ב-Dataflow בדף הזה.

המשך

ב-Apache Beam SDK ל-Go אי אפשר להגדיר אפשרויות של צינור עיבוד נתונים באופן פרוגרמטי באמצעות PipelineOptions. צריך להשתמש בארגומנטים של שורת הפקודה ב-Go.

לדוגמה, ראו את הקטע דוגמה להפעלה ב-Dataflow בדף הזה.

הגדרת אפשרויות של צינור עיבוד נתונים בשורת הפקודה

אפשר להגדיר את האפשרויות של צינור עיבוד הנתונים באמצעות ארגומנטים של שורת הפקודה.

Java

התחביר בדוגמה הבאה הוא מצינור עיבוד הנתונים WordCount במדריך ל-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"

מחליפים את מה שכתוב בשדות הבאים:

  • ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud
  • ‫BUCKET_NAME: השם של הקטגוריה של Cloud Storage
  • ‫REGION: אזור Dataflow, us-central1

Python

התחביר לדוגמה הבא הוא מצינור עיבוד הנתונים WordCount במדריך 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/

מחליפים את מה שכתוב בשדות הבאים:

  • ‫DATAFLOW_REGION: האזור שבו רוצים לפרוס את משימת Dataflow, לדוגמה europe-west1

    הדגל --region מבטל את ברירת המחדל של האזור שמוגדר בשרת המטא-נתונים, בלקוח המקומי או במשתני הסביבה.

  • ‫STORAGE_BUCKET: שם הקטגוריה ב-Cloud Storage

  • ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud

המשך

התחביר לדוגמה הבא הוא מצינור עיבוד הנתונים WordCount במדריך של 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/

מחליפים את מה שכתוב בשדות הבאים:

  • ‫BUCKET_NAME: שם הקטגוריה של Cloud Storage

  • ‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud

  • ‫DATAFLOW_REGION: האזור שבו רוצים לפרוס את משימת Dataflow. לדוגמה, europe-west1. הדגל --region מבטל את ברירת המחדל של האזור שמוגדר בשרת המטא-נתונים, בלקוח המקומי או במשתני הסביבה.

הגדרת אפשרויות ניסיוניות לצינור עיבוד הנתונים

ב-Java, ב-Python וב-Go SDK, experiments אפשרות צינור עיבוד הנתונים מפעילה תכונות ניסיוניות או תכונות טרום-GA של Dataflow.

הגדרה פרוגרמטית

כדי להגדיר את האפשרות experiments באופן פרוגרמטי, משתמשים בתחביר הבא.

Java

באובייקט PipelineOptions, כוללים את האפשרות experiments באמצעות התחביר הבא. בדוגמה הזו, גודל דיסק האתחול מוגדר ל-80GB באמצעות דגל הניסוי.

options.setExperiments("streaming_boot_disk_size_gb=80")

בקטע דוגמה להפעלה ב-Dataflow שבדף הזה מופיעה דוגמה שמראה איך ליצור את האובייקט PipelineOptions.

Python

באובייקט PipelineOptions, כוללים את האפשרות experiments באמצעות התחביר הבא. בדוגמה הזו, גודל דיסק האתחול מוגדר ל-80GB באמצעות דגל הניסוי.

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

בקטע דוגמה להפעלה ב-Dataflow שבדף הזה מופיעה דוגמה שמראה איך ליצור את האובייקט PipelineOptions.

המשך

ב-Apache Beam SDK ל-Go אי אפשר להגדיר אפשרויות של צינור עיבוד נתונים באופן פרוגרמטי באמצעות PipelineOptions. צריך להשתמש בארגומנטים של שורת הפקודה ב-Go.

הגדרה בשורת הפקודה

כדי להגדיר את האפשרות experiments בשורת הפקודה, משתמשים בתחביר הבא.

Java

בדוגמה הזו, גודל דיסק האתחול מוגדר ל-80GB באמצעות דגל הניסוי.

--experiments=streaming_boot_disk_size_gb=80

Python

בדוגמה הזו, גודל דיסק האתחול מוגדר ל-80GB באמצעות דגל הניסוי.

--experiments=streaming_boot_disk_size_gb=80

המשך

בדוגמה הזו, גודל דיסק האתחול מוגדר ל-80GB באמצעות דגל הניסוי.

--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[,...]

גישה לאובייקט של אפשרויות צינור עיבוד הנתונים

כשיוצרים את האובייקט Pipeline בתוכנית Apache Beam, מעבירים את PipelineOptions. כששירות Dataflow מריץ את צינור עיבוד הנתונים, הוא שולח עותק של PipelineOptions לכל worker.

Java

כדי לגשת ל-PipelineOptions בתוך מופע DoFn של טרנספורמציה ב-ParDo, משתמשים ב-method‏ ProcessContext.getPipelineOptions.

Python

התכונה הזו לא נתמכת ב-Apache Beam SDK for Python.

המשך

אפשר לגשת לאפשרויות של צינור עיבוד הנתונים באמצעות beam.PipelineOptions.

הפעלה ב-Dataflow

מריצים את המשימה במשאבים מנוהלים של Google Cloud באמצעות שירות ה-runner של Dataflow. כשמריצים את צינור עיבוד הנתונים באמצעות Dataflow, נוצרת משימת Dataflow שמשתמשת במשאבים של Compute Engine ו-Cloud Storage בפרויקט ב- Google Cloud. למידע על ההרשאות ב-Dataflow, קראו את המאמר אבטחה והרשאות ב-Dataflow.

משימות Dataflow משתמשות ב-Cloud Storage כדי לאחסן קבצים זמניים במהלך ההפעלה של צינור עיבוד הנתונים. כדי להימנע מעלויות אחסון מיותרות, כדאי להשבית את התכונה "מחיקה זמנית" בקטגוריות שמשמשות את משימות Dataflow לאחסון זמני. מידע נוסף זמין במאמר השבתת המחיקה הרכה.

הגדרת האפשרויות הנדרשות

כדי להריץ את צינור עיבוד הנתונים באמצעות Dataflow, מגדירים את האפשרויות הבאות:

Java

  • ‫project: מזהה הפרויקט ב- Google Cloud .
  • ‫runner: מריץ צינור עיבוד הנתונים שמפעיל את צינור עיבוד הנתונים. כדי להפעיל אתGoogle Cloud , צריך להגדיר את DataflowRunner.

  • ‫gcpTempLocation: נתיב ב-Cloud Storage שבו Dataflow יאחסן את רוב הקבצים הזמניים. הקטגוריה שציינתם כבר צריכה להיות קיימת.

    אם לא מציינים את gcpTempLocation, ‏Dataflow משתמש בערך של האפשרות tempLocation. אם לא תציינו אף אחת מהאפשרויות האלה, Dataflow ייצור קטגוריה חדשה ב-Cloud Storage.

Python

  • ‫project: מזהה הפרויקט ב- Google Cloud .
  • ‫region: האזור של משימת Dataflow.
  • ‫runner: מריץ צינור עיבוד הנתונים שמפעיל את צינור עיבוד הנתונים. כדי להפעיל אתGoogle Cloud , צריך להגדיר את DataflowRunner.
  • ‫temp_location: נתיב ב-Cloud Storage שבו Dataflow יאחסן קובצי משימות זמניים שנוצרו במהלך הפעלת צינור עיבוד הנתונים.

המשך

  • ‫project: מזהה הפרויקט ב- Google Cloud .
  • ‫region: האזור של משימת Dataflow.
  • ‫runner: מריץ צינור עיבוד הנתונים שמפעיל את צינור עיבוד הנתונים. כדי להפעיל אתGoogle Cloud , צריך להגדיר את dataflow.
  • ‫staging_location: נתיב ב-Cloud Storage שבו Dataflow יאחסן קובצי משימות זמניים שנוצרו במהלך הפעלת צינור עיבוד הנתונים.

הגדרת אפשרויות צינור עיבוד נתונים באופן פרוגרמטי

בדוגמת הקוד הבאה אפשר לראות איך לבנות צינור עיבוד נתונים על ידי הגדרה פרוגרמטית של ה-runner ואפשרויות נדרשות אחרות להפעלת צינור עיבוד הנתונים באמצעות 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.

המשך

ערכת ה-SDK של Apache Beam ל-Go משתמשת בארגומנטים של שורת הפקודה של 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

משתמשים במודול argparse של Python כדי לנתח את האפשרויות בשורת הפקודה.

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

המשך

משתמשים בחבילת flag של Go כדי לנתח את האפשרויות של שורת הפקודה. צריך לנתח את האפשרויות לפני שקוראים ל-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()

אחרי שיוצרים את צינור עיבוד הנתונים, מציינים את כל פעולות הקריאה, הטרנספורמציה והכתיבה של צינור עיבוד הנתונים, ואז מריצים אותו.

הגדרת אפשרויות נוספות לצינור עיבוד הנתונים

בנוסף להגדרת האפשרויות כשמריצים פייפליינים ישירות באמצעות ערכות ה-SDK של Apache Beam, אפשר להגדיר את אפשרויות הפייפליין וההרצה כשמפעילים משימות באמצעות ה-CLI של gcloud (למשל כשמריצים תבניות או פייפליינים של YAML) או באמצעות Dataflow API בארכיטקטורת REST.

הערה: לרשימה המלאה של אפשרויות ודגלים של צינור עיבוד נתונים, ראו אפשרויות של צינור עיבוד נתונים.

בכרטיסיות הבאות מוסבר איך להגדיר אפשרויות נפוצות של וורקרים, רשתות והפעלה באמצעות ה-CLI של gcloud ו-API בארכיטקטורת REST.

gcloud

כשמפעילים משימות או תבניות באמצעות פקודות כמו gcloud dataflow jobs run,‏ gcloud dataflow flex-template run או gcloud dataflow yaml run, משתמשים בדגלים של שורת הפקודה כדי לציין את הגדרות הביצוע וה-worker.

מידע נוסף על הגדרת האפשרויות של תבניות Flex זמין במאמר הגדרת אפשרויות צינור עיבוד נתונים לתבניות Flex.

במאמרי העזרה של gcloud dataflow יש רשימה מלאה של דגלים לכל פקודה ב-CLI של gcloud.

API

כדי לציין אפשרויות הרצה ו-worker כשמפעילים משימה או תבנית באמצעות Dataflow API בארכיטקטורת REST, עיינו בדפי העזרה הבאים:

מצבי הפעלה של אמצעי בקרה

כשתוכנית Apache Beam מריצה צינור עיבוד נתונים בשירות כמו Dataflow, התוכנית יכולה להריץ את צינור עיבוד הנתונים באופן אסינכרוני, או לחסום עד להשלמת צינור עיבוד הנתונים. כדי לשנות את ההתנהגות הזו, צריך לפעול לפי ההנחיות הבאות.

Java

כשתוכנית Apache Beam Java מריצה צינור עיבוד נתונים בשירות כמו Dataflow, בדרך כלל היא מופעלת באופן אסינכרוני. כדי להריץ צינור עיבוד נתונים ולהמתין עד שהמשימה תסתיים, מגדירים את DataflowRunner כמריץ צינור עיבוד הנתונים וקוראים באופן מפורש ל-pipeline.run().waitUntilFinish().

כשמשתמשים ב-DataflowRunner ומפעילים את waitUntilFinish() באובייקט PipelineResult שמוחזר מ-pipeline.run(), צינור עיבוד הנתונים פועל ב- Google Cloud , אבל הקוד המקומי ממתין עד שהמשימה בענן תסתיים ותחזיר את האובייקט הסופי DataflowPipelineJob. בזמן שהמשימה פועלת, שירות Dataflow מדפיס עדכונים על סטטוס המשימה והודעות במסוף בזמן ההמתנה.

Python

כשתוכנית Apache Beam Python מריצה צינור עיבוד נתונים בשירות כמו Dataflow, בדרך כלל היא מופעלת באופן אסינכרוני. כדי לחסום עד להשלמת צינור עיבוד הנתונים, משתמשים ב-method‏ wait_until_finish() של האובייקט PipelineResult, שמוחזר מה-method‏ run() של ה-runner.

המשך

כשתוכנית Apache Beam Go מריצה צינור עיבוד נתונים ב-Dataflow, היא פועלת באופן סינכרוני כברירת מחדל וחוסמת עד להשלמת צינור עיבוד הנתונים. אם אתם לא רוצים לחסום, יש שתי אפשרויות:

  1. מפעילים את המשימה ב-Go routine.

    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. משתמשים בדגל שורת הפקודה --async, שנמצא בחבילה jobopts.

כדי לראות את פרטי ההפעלה, לעקוב אחרי ההתקדמות ולבדוק את סטטוס ההשלמה של המשימה, משתמשים בממשק המעקב של Dataflow או בממשק שורת הפקודה של Dataflow.

שימוש במקורות סטרימינג

Java

אם צינור עיבוד הנתונים קורא ממקור נתונים לא מוגבל, כמו Pub/Sub, הוא פועל באופן אוטומטי במצב סטרימינג.

Python

אם צינור עיבוד הנתונים משתמש במקור נתונים לא מוגבל, כמו Pub/Sub, צריך להגדיר את האפשרות streaming כ-true.

המשך

אם צינור עיבוד הנתונים קורא ממקור נתונים לא מוגבל, כמו Pub/Sub, הוא פועל באופן אוטומטי במצב סטרימינג.

כברירת מחדל, משימות סטרימינג משתמשות בסוג מכונה של Compute Engine עם 2 מעבדי vCPU או יותר.

הפעלה מקומית

במקום להריץ את צינור עיבוד הנתונים במשאבי ענן מנוהלים, אתם יכולים להריץ אותו באופן מקומי. להרצה מקומית יש יתרונות מסוימים לבדיקה, לניפוי באגים או להרצה של צינור עיבוד הנתונים על קבוצות נתונים קטנות. לדוגמה, הפעלה מקומית מסירה את התלות בשירות 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))

המשך

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

כדי להוסיף אפשרויות משלכם, השתמשו ב-method‏ add_argument() (שפועלת בדיוק כמו המודול argparse הרגיל של Python), כמו בדוגמה הבאה:

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, כמו בדוגמה הבאה:

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

המשך

מגדירים את התיאור ואת ערך ברירת המחדל באופן הבא:

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