כתיבה מ-Dataflow ל-BigQuery

במאמר הזה מוסבר איך לכתוב נתונים מ-Dataflow ל-BigQuery.

סקירה כללית

ברוב תרחישי השימוש, כדאי להשתמש ב-Managed I/O כדי לכתוב ל-BigQuery. קלט/פלט מנוהל מספק תכונות כמו שדרוגים אוטומטיים וממשק API עקבי להגדרות אישיות. כשכותבים ל-BigQuery, קלט/פלט מנוהל בוחר באופן אוטומטי את שיטת הכתיבה הטובה ביותר למשימות באצווה או בסטרימינג.

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

ביצועים

בטבלה הבאה מוצגים מדדי ביצועים לעומסי עבודה שונים. עומסי העבודה האלה הופעלו ב-worker אחד של e2-standard2, באמצעות Apache Beam SDK 2.49.0 ל-Java. הם לא השתמשו ב-Runner v2.

‫100 מיליון רשומות | 1KB | עמודה אחת	תפוקה (בייטים)	תפוקה (רכיבים)
כתיבה באחסון	‫55MB לשנייה	‫54,000 רכיבים לשנייה
טעינת Avro	78 MBps	77,000 רכיבים לשנייה
טעינת JSON	54 MBps	‫53,000 רכיבים לשנייה

המדדים האלה מבוססים על צינורות עיבוד נתונים פשוטים של אצווה. הם נועדו להשוות בין הביצועים של מחברי קלט/פלט, ולא בהכרח מייצגים צינורות עיבוד נתונים בעולם האמיתי. הביצועים של צינור עיבוד הנתונים ב-Dataflow הם מורכבים, והם פונקציה של סוג ה-VM, הנתונים שעוברים עיבוד, הביצועים של מקורות וכיורים חיצוניים וקוד המשתמש. המדדים מבוססים על הפעלת Java SDK, והם לא מייצגים את מאפייני הביצועים של ערכות SDK בשפות אחרות. מידע נוסף זמין במאמר ביצועי קלט/פלט של Beam.

שימוש במחבר BigQueryIO

מחבר הקלט/פלט של BigQuery תומך בשיטות הבאות לכתיבה ב-BigQuery:

• STORAGE_WRITE_API. במצב הזה, המחבר מבצע כתיבה ישירה לאחסון ב-BigQuery באמצעות BigQuery Storage Write API (gRPC). ‫Storage Write API ‏ (gRPC) משלב הטמעת עדכונים בזמן אמת וטעינת נתונים בכמות גדולה בממשק API אחד עם ביצועים משופרים. המצב הזה מבטיח סמנטיקה של פעם אחת בדיוק.
• STORAGE_API_AT_LEAST_ONCE. גם במצב הזה נעשה שימוש ב-Storage Write API ‏ (gRPC), אבל הוא מספק סמנטיקה של "לפחות פעם אחת". במצב הזה, זמן האחזור קצר יותר ברוב צינורות עיבוד הנתונים. עם זאת, יכול להיות שיהיו כתיבות כפולות.
• FILE_LOADS. במצב הזה, המחבר כותב את נתוני הקלט לקובצי Staging ב-Cloud Storage. לאחר מכן, הוא מריץ משימת טעינה של BigQuery כדי לטעון את הנתונים ל-BigQuery. זהו מצב ברירת המחדל של PCollections מוגבלים, שבדרך כלל נמצאים בצינורות עיבוד נתונים באצווה.
• STREAMING_INSERTS. במצב הזה, המחבר משתמש ב-API הקודם להעברה בסטרימינג. במצב הזה מתבצעת מניעת כפילויות במידת האפשר, ולכן יכול להיות שיהיו כמה כפילויות. זהו מצב ברירת המחדל של PCollections ללא הגבלה, אבל הוא לא מומלץ לפרויקטים חדשים.

כשבוחרים שיטת כתיבה, כדאי להביא בחשבון את הנקודות הבאות:

• במשימות סטרימינג, כדאי להשתמש ב-STORAGE_WRITE_API או ב-STORAGE_API_AT_LEAST_ONCE, כי המצבים האלה כותבים ישירות לאחסון ב-BigQuery, בלי להשתמש בקובצי ביניים.
• אם מריצים את צינור עיבוד הנתונים באמצעות מצב סטרימינג של לפחות פעם אחת, צריך להגדיר את מצב הכתיבה ל-STORAGE_API_AT_LEAST_ONCE. ההגדרה הזו יעילה יותר ותואמת לסמנטיקה של מצב הסטרימינג "לפחות פעם אחת". שיטת הכתיבה FILE_LOADS לא תואמת למצב סטרימינג של "לפחות פעם אחת".
• לטעינת קבצים ול-Storage Write API ‏ (gRPC) יש מכסות ומגבלות שונות.
• משימות טעינה משתמשות במאגר המשותף של יחידות הקיבולת של BigQuery או ביחידות קיבולת שמורות. כדי להשתמש במשבצות שהוקצו, מריצים את משימת הטעינה בפרויקט עם הקצאת הזמנה מסוג PIPELINE. משימות טעינה לא עולות כסף אם משתמשים במאגר המשותף של יחידות הקיבולת של BigQuery. עם זאת, BigQuery לא מתחייבת לגבי הקיבולת הזמינה של המאגר המשותף. מידע נוסף מופיע במאמר מבוא להזמנות.

מקביליות

• ב-FILE_LOADS וב-STORAGE_WRITE_API בפייפליינים זורמים, המחבר מפצל את הנתונים למספר קבצים או סטרימינג. באופן כללי, מומלץ להפעיל את withAutoSharding כדי להפעיל חלוקה אוטומטית לפלחים.

• ב-FILE_LOADS בצינורות עיבוד נתונים באצווה, המחבר כותב נתונים לקבצים מחולקים למחיצות, שנטענים לאחר מכן ל-BigQuery במקביל.

• ב-STORAGE_WRITE_API בצינורות עיבוד נתונים באצווה, כל worker יוצר לפחות זרם אחד לכתיבה ב-BigQuery, בהתאם למספר הכולל של פיצולי הנתונים.

• ל-STORAGE_API_AT_LEAST_ONCE יש זרם כתיבה אחד כברירת מחדל. כמה וורקרים מוסיפים נתונים לסטרימינג הזה.

שיטות מומלצות

• ל-Storage Write API ‏ (gRPC) יש מגבלות מכסה. המחבר מטפל במגבלות האלה ברוב צינורות עיבוד הנתונים. עם זאת, בתרחישים מסוימים יכול להיות שכל הסטרימינג הזמין ב-Storage Write API ‏ (gRPC) ינוצל. לדוגמה, הבעיה הזו יכולה לקרות בצינור עיבוד נתונים שמשתמש בפיצול אוטומטי ובהתאמה אוטומטית לעומס עם מספר גדול של יעדים, במיוחד במשימות ארוכות עם עומסי עבודה משתנים מאוד. אם הבעיה הזו מתרחשת, כדאי להשתמש ב-STORAGE_WRITE_API_AT_LEAST_ONCE כדי למנוע אותה.

• אתם יכולים להשתמש בGoogle Cloud מדדים כדי לעקוב אחרי השימוש במכסות של Storage Write API ‏ (gRPC).

• כשמשתמשים בטעינת קבצים, בדרך כלל Avro עדיף על JSON. כדי להשתמש ב-Avro, מפעילים את withAvroFormatFunction.

• כברירת מחדל, משימות הטעינה פועלות באותו פרויקט שבו פועלת משימת Dataflow. כדי לציין פרויקט אחר, צריך להפעיל את withLoadJobProjectId.

• כשמשתמשים ב-Java SDK, כדאי ליצור מחלקה שמייצגת את הסכימה של הטבלה ב-BigQuery. ואז קוראים ל-useBeamSchema בצינור עיבוד הנתונים כדי להמיר באופן אוטומטי בין סוגי Apache Beam Row ו-BigQuery TableRow. דוגמה למחלקת סכימה זמינה בקישור ExampleModel.java.

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

• כברירת מחדל, BigQueryIO משתמש בהגדרות של Storage Write API ‏ (gRPC) שמתאימות לרוב צינורות עיבוד הנתונים. עם זאת, אם אתם נתקלים בבעיות בביצועים, אתם יכולים להגדיר אפשרויות של צינור עיבוד נתונים כדי לשנות את ההגדרות האלה. מידע נוסף מופיע במאמר כוונון Storage Write API ‏ (gRPC) במסמכי התיעוד של Apache Beam.

פייפליינים זורמים

ההמלצות הבאות רלוונטיות לפייפליינים זורמים.

• בפייפליינים זורמים, מומלץ להשתמש ב-Storage Write API ‏ (gRPC) (STORAGE_WRITE_API או STORAGE_API_AT_LEAST_ONCE).

• אפשר להשתמש בטעינת קבצים בפייפליין זורם, אבל לגישה הזו יש חסרונות:

  • כדי לכתוב את הקבצים, צריך להשתמש בחלונות. אי אפשר להשתמש בחלון הגלובלי.
  • כשמשתמשים במאגר ה-slots המשותף, המערכת של BigQuery טוענת קבצים על בסיס מאמץ מרבי. יכול להיות עיכוב משמעותי בין הזמן שבו רשומה נכתבת לבין הזמן שבו היא זמינה ב-BigQuery.
  • אם משימת טעינה נכשלת, למשל בגלל נתונים פגומים או אי-התאמה לסכימה, כל צינור עיבוד הנתונים נכשל.

• כדאי להשתמש ב-STORAGE_WRITE_API_AT_LEAST_ONCE כשאפשר. היא יכולה לגרום לכתיבת רשומות כפולות ב-BigQuery, אבל היא זולה יותר וניתנת להתאמה לעומס (scaling) יותר מ-STORAGE_WRITE_API.

• באופן כללי, מומלץ להימנע משימוש ב-STREAMING_INSERTS. הזנת זרם נתונים יקרה יותר מ-Storage Write API ‏ (gRPC), הביצועים שלה פחות טובים והיא משתמשת בביטול כפילויות במידת האפשר, מה שעלול לגרום לכפילויות מסוימות.

• פיצול נתונים יכול לשפר את הביצועים בפייפליינים זורמים. ברוב צינורות עיבוד הנתונים, חלוקה אוטומטית ל-shards היא נקודת התחלה טובה. עם זאת, אפשר לשנות את חלוקת הנתונים לפלחים באופן הבא:

  • ב-STORAGE_WRITE_API, צריך להפעיל את withNumStorageWriteApiStreams כדי להגדיר את מספר ה-write streams.
  • ב-FILE_LOADS, מפעילים את withNumFileShards כדי להגדיר את מספר חלקי הקובץ.

• אם אתם משתמשים בהזנת זרם נתונים, מומלץ להגדיר את retryTransientErrors בתור מדיניות הניסיון החוזר.

צינורות עיבוד נתונים באצווה

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

• ברוב צינורות עיבוד הנתונים של אצווה גדולה, מומלץ לנסות קודם את FILE_LOADS. בצינור עיבוד נתונים באצווה אפשר להשתמש ב-STORAGE_WRITE_API, אבל סביר להניח שהוא יחרוג ממגבלות המכסות בקנה מידה גדול (יותר מ-1,000 מעבדי vCPU) או אם פועלים צינורות עיבוד נתונים בו-זמנית. ‫Apache Beam לא מגביל את המספר המקסימלי של זרמי כתיבה למשימות באצווה של STORAGE_WRITE_API, ולכן המשימה מגיעה בסופו של דבר למגבלות של BigQuery Storage API.

• כשמשתמשים ב-FILE_LOADS, יכול להיות שתנצלו את כל מאגר יחידות הקיבולת המשותפות של BigQuery או את כל מאגר יחידות הקיבולת השמורות שלכם. אם נתקלתם בכשל כזה, נסו את הגישות הבאות:

  • מפחיתים את המספר המקסימלי של ה-workers או את הגודל שלהם במשימה.
  • לרכוש עוד משבצות שמורות.
  • כדאי להשתמש ב-STORAGE_WRITE_API.

• בצינורות עיבוד נתונים קטנים עד בינוניים (פחות מ-1,000 מעבדי vCPU), כדאי להשתמש ב-STORAGE_WRITE_API. במשימות קטנות כאלה, מומלץ להשתמש ב-STORAGE_WRITE_API אם רוצים תור להודעות שלא נמסרו או כשמאגר המשבצות המשותף של FILE_LOADS לא מספיק.

• אם אתם יכולים להסתדר עם נתונים כפולים, כדאי להשתמש ב-STORAGE_WRITE_API_AT_LEAST_ONCE. במצב הזה יכול להיות שרשומות כפולות ייכתבו ב-BigQuery, אבל הוא עשוי להיות זול יותר מהאפשרות STORAGE_WRITE_API.

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

טיפול בשגיאות ברמת השורה

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

ב-Storage Write API ‏ (gRPC), כל השורות שלא ניתן לכתוב ממוקמות ב-PCollection נפרד. כדי לקבל את האוסף הזה, צריך להפעיל את getFailedStorageApiInserts באובייקט WriteResult. במאמר הזה מוסבר איך להעביר נתונים בסטרימינג ל-BigQuery.

מומלץ לשלוח את השגיאות לתור או לטבלה של הודעות שלא נמסרו, כדי לעבד אותן מאוחר יותר. מידע נוסף על התבנית הזו זמין במאמר תבנית של הודעות שלא נמסרו ב-BigQueryIO.

ב-FILE_LOADS, אם מתרחשת שגיאה בזמן טעינת הנתונים, משימת הטעינה נכשלת והצינור זורק חריגה בזמן ריצה. אפשר לראות את השגיאה ביומנים של Dataflow או בהיסטוריית המשימות של BigQuery. מחבר הקלט/פלט לא מחזיר מידע על שורות ספציפיות שנכשלו.

למידע נוסף על פתרון שגיאות, ראו שגיאות במחבר ל-BigQuery.

דוגמאות

בדוגמאות הבאות מוסבר איך להשתמש ב-Dataflow כדי לכתוב ב-BigQuery. בדוגמאות האלה נעשה שימוש במחבר BigQueryIO.

כתיבה לטבלה קיימת

בדוגמה הבאה נוצר צינור עיבוד נתונים באצווה שכותב PCollection<MyData> ל-BigQuery, כאשר MyData הוא סוג נתונים בהתאמה אישית.

השיטה BigQueryIO.write() מחזירה סוג של BigQueryIO.Write<T>, שמשמש להגדרת פעולת הכתיבה. מידע נוסף זמין במאמר כתיבה לטבלה במסמכי התיעוד של Apache Beam. דוגמת הקוד הזו כותבת לטבלה קיימת (CREATE_NEVER) ומוסיפה את השורות החדשות לטבלה (WRITE_APPEND).

Java

כדי לבצע אימות ב-Dataflow, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

import com.google.api.services.bigquery.model.TableRow;
import java.util.Arrays;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.extensions.avro.coders.AvroCoder;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;

public class BigQueryWrite {
  // A custom datatype for the source data.
  @DefaultCoder(AvroCoder.class)
  public static class MyData {
    public String name;
    public Long age;

    public MyData() {}

    public MyData(String name, Long age) {
      this.name = name;
      this.age = age;
    }
  }

  public static void main(String[] args) {
    // Example source data.
    final List<MyData> data = Arrays.asList(
        new MyData("Alice", 40L),
        new MyData("Bob", 30L),
        new MyData("Charlie", 20L)
    );

    // Parse the pipeline options passed into the application. Example:
    //   --projectId=$PROJECT_ID --datasetName=$DATASET_NAME --tableName=$TABLE_NAME
    // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    PipelineOptionsFactory.register(ExamplePipelineOptions.class);
    ExamplePipelineOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(ExamplePipelineOptions.class);

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        // Create an in-memory PCollection of MyData objects.
        .apply(Create.of(data))
        // Write the data to an exiting BigQuery table.
        .apply(BigQueryIO.<MyData>write()
            .to(String.format("%s:%s.%s",
                options.getProjectId(),
                options.getDatasetName(),
                options.getTableName()))
            .withFormatFunction(
                (MyData x) -> new TableRow().set("user_name", x.name).set("age", x.age))
            .withCreateDisposition(CreateDisposition.CREATE_NEVER)
            .withWriteDisposition(WriteDisposition.WRITE_APPEND)
            .withMethod(Write.Method.STORAGE_WRITE_API));
    pipeline.run().waitUntilFinish();
  }
}

כתיבה בטבלה חדשה או קיימת

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

Java

כדי לבצע אימות ב-Dataflow, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

import com.google.api.services.bigquery.model.TableFieldSchema;
import com.google.api.services.bigquery.model.TableRow;
import com.google.api.services.bigquery.model.TableSchema;
import java.util.Arrays;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.extensions.avro.coders.AvroCoder;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.Create;

public class BigQueryWriteWithSchema {
  // A custom datatype for the source data.
  @DefaultCoder(AvroCoder.class)
  public static class MyData {
    public String name;
    public Long age;

    public MyData() {}

    public MyData(String name, Long age) {
      this.name = name;
      this.age = age;
    }
  }

  public static void main(String[] args) {
    // Example source data.
    final List<MyData> data = Arrays.asList(
        new MyData("Alice", 40L),
        new MyData("Bob", 30L),
        new MyData("Charlie", 20L)
    );

    // Define a table schema. A schema is required for write disposition CREATE_IF_NEEDED.
    TableSchema schema = new TableSchema()
        .setFields(
            Arrays.asList(
                new TableFieldSchema()
                    .setName("user_name")
                    .setType("STRING")
                    .setMode("REQUIRED"),
                new TableFieldSchema()
                    .setName("age")
                    .setType("INT64") // Defaults to NULLABLE
            )
        );

    // Parse the pipeline options passed into the application. Example:
    //   --projectId=$PROJECT_ID --datasetName=$DATASET_NAME --tableName=$TABLE_NAME
    // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    PipelineOptionsFactory.register(ExamplePipelineOptions.class);
    ExamplePipelineOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(ExamplePipelineOptions.class);

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        // Create an in-memory PCollection of MyData objects.
        .apply(Create.of(data))
        // Write the data to a new or existing BigQuery table.
        .apply(BigQueryIO.<MyData>write()
            .to(String.format("%s:%s.%s",
                options.getProjectId(),
                options.getDatasetName(),
                options.getTableName()))
            .withFormatFunction(
                (MyData x) -> new TableRow().set("user_name", x.name).set("age", x.age))
            .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
            .withSchema(schema)
            .withMethod(Write.Method.STORAGE_WRITE_API)
        );
    pipeline.run().waitUntilFinish();
  }
}

העברת נתונים בסטרימינג ל-BigQuery

בדוגמה הבאה מוסבר איך להעביר נתונים בסטרימינג באמצעות סמנטיקה של פעם אחת בדיוק, על ידי הגדרת מצב הכתיבה ל-STORAGE_WRITE_API

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

Java

כדי לבצע אימות ב-Dataflow, צריך להגדיר את Application Default Credentials. למידע נוסף, ראו הגדרת אימות לסביבת פיתוח מקומית.

import com.google.api.services.bigquery.model.TableRow;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.CreateDisposition;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write.WriteDisposition;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.testing.TestStream;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TimestampedValue;
import org.apache.beam.sdk.values.TypeDescriptor;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.joda.time.Duration;
import org.joda.time.Instant;

public class BigQueryStreamExactlyOnce {
  // Create a PTransform that sends simulated streaming data. In a real application, the data
  // source would be an external source, such as Pub/Sub.
  private static TestStream<String> createEventSource() {
    Instant startTime = new Instant(0);
    return TestStream.create(StringUtf8Coder.of())
        .advanceWatermarkTo(startTime)
        .addElements(
            TimestampedValue.of("Alice,20", startTime),
            TimestampedValue.of("Bob,30",
                startTime.plus(Duration.standardSeconds(1))),
            TimestampedValue.of("Charles,40",
                startTime.plus(Duration.standardSeconds(2))),
            TimestampedValue.of("Dylan,Invalid value",
                startTime.plus(Duration.standardSeconds(2))))
        .advanceWatermarkToInfinity();
  }

  public static PipelineResult main(String[] args) {
    // Parse the pipeline options passed into the application. Example:
    //   --projectId=$PROJECT_ID --datasetName=$DATASET_NAME --tableName=$TABLE_NAME
    // For more information, see https://beam.apache.org/documentation/programming-guide/#configuring-pipeline-options
    PipelineOptionsFactory.register(ExamplePipelineOptions.class);
    ExamplePipelineOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(ExamplePipelineOptions.class);
    options.setStreaming(true);

    // Create a pipeline and apply transforms.
    Pipeline pipeline = Pipeline.create(options);
    pipeline
        // Add a streaming data source.
        .apply(createEventSource())
        // Map the event data into TableRow objects.
        .apply(MapElements
            .into(TypeDescriptor.of(TableRow.class))
            .via((String x) -> {
              String[] columns = x.split(",");
              return new TableRow().set("user_name", columns[0]).set("age", columns[1]);
            }))
        // Write the rows to BigQuery
        .apply(BigQueryIO.writeTableRows()
            .to(String.format("%s:%s.%s",
                options.getProjectId(),
                options.getDatasetName(),
                options.getTableName()))
            .withCreateDisposition(CreateDisposition.CREATE_NEVER)
            .withWriteDisposition(WriteDisposition.WRITE_APPEND)
            .withMethod(Write.Method.STORAGE_WRITE_API)
            // For exactly-once processing, set the triggering frequency.
            .withTriggeringFrequency(Duration.standardSeconds(5)))
        // Get the collection of write errors.
        .getFailedStorageApiInserts()
        .apply(MapElements.into(TypeDescriptors.strings())
            // Process each error. In production systems, it's useful to write the errors to
            // another destination, such as a dead-letter table or queue.
            .via(
                x -> {
                  System.out.println("Failed insert: " + x.getErrorMessage());
                  System.out.println("Row: " + x.getRow());
                  return "";
                }));
    return pipeline.run();
  }
}

המאמרים הבאים

• מידע נוסף על קלט/פלט מנוהל
• מידע ולמידה על שיטות מומלצות להעברת נתונים מ-Pub/Sub ל-BigQuery
• להבין את מאפייני הביצועים של צינורות עיבוד נתונים מ-Pub/Sub ל-BigQuery