קריאה מ-Cloud Storage אל Dataflow

כדי לקרוא נתונים מ-Cloud Storage ל-Dataflow, משתמשים במחבר הקלט/פלט של Apache Beam‏ TextIO או AvroIO.

כוללים את התלות בספריית Google Cloud

כדי להשתמש במחבר TextIO או AvroIO עם Cloud Storage, צריך לכלול את התלות הבאה. הספרייה הזו מספקת רכיב handler של סכימה לשמות קבצים מסוג "gs://".

Java 

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
  <version>${beam.version}</version>
</dependency>

Python 

apache-beam[gcp]==VERSION

המשך 

import _ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/gcs"

מידע נוסף זמין במאמר בנושא התקנת Apache Beam SDK.

הפעלת gRPC במחבר קלט/פלט של Apache Beam ב-Dataflow

אפשר להתחבר ל-Cloud Storage באמצעות gRPC דרך מחבר הקלט/פלט של Apache Beam ב-Dataflow. ‏gRPC הוא פריימוורק קוד פתוח של קריאות לפרוצדורות מרוחקות (RPC) עם ביצועים גבוהים שפותח על ידי Google, ואפשר להשתמש בו כדי ליצור אינטראקציה עם Cloud Storage.

כדי להאיץ את בקשות הקריאה של משימת Dataflow ל-Cloud Storage, אפשר להפעיל את מחבר הקלט/פלט של Apache Beam ב-Dataflow כדי להשתמש ב-gRPC.

שורת הפקודה

  1. חשוב לוודא שאתם משתמשים בגרסה 2.55.0 ואילך של Apache Beam SDK.
  2. כדי להריץ משימת Dataflow, משתמשים באפשרות --additional-experiments=use_grpc_for_gcs pipeline. מידע על האפשרויות השונות של צינורות העברת נתונים מופיע במאמר בנושא דגלים אופציונליים.

‫Apache Beam SDK

  1. חשוב לוודא שאתם משתמשים בגרסה 2.55.0 ואילך של Apache Beam SDK.
  2. כדי להריץ משימת Dataflow, משתמשים באפשרות --experiments=use_grpc_for_gcs pipeline. מידע על האפשרויות השונות של צינורות העברת נתונים מופיע במאמר בנושא אפשרויות בסיסיות.

אפשר להגדיר מחבר קלט/פלט של Apache Beam ב-Dataflow כדי ליצור מדדים שקשורים ל-gRPC ב-Cloud Monitoring. המדדים שקשורים ל-gRPC יכולים לעזור לכם:

  • מעקב אחרי הביצועים של בקשות gRPC ל-Cloud Storage ושיפור שלהם.
  • לפתור בעיות ולנפות באגים.
  • קבלת תובנות לגבי השימוש באפליקציה וההתנהגות שלה.

מידע על הגדרת מחבר Apache Beam I/O ב-Dataflow כדי ליצור מדדים שקשורים ל-gRPC זמין במאמר שימוש במדדים בצד הלקוח. אם איסוף מדדים לא נחוץ לתרחיש השימוש שלכם, אתם יכולים לבטל את ההסכמה לאיסוף מדדים. הוראות מפורטות מופיעות במאמר ביטול ההסכמה לשיתוף מדדים בצד הלקוח.

מקביליות

מחברי TextIO ו-AvroIO תומכים בשתי רמות של מקביליות:

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

ביצועים

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

‫100 מיליון רשומות | 1KB | עמודה אחת תפוקה (בייטים) תפוקה (אלמנטים)
קריאה ‫320 MBps ‫320,000 רכיבים לשנייה

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

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

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

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

דוגמה

בדוגמה הבאה מוצג אופן הקריאה מ-Cloud Storage.

Java

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

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
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.MapElements;
import org.apache.beam.sdk.values.TypeDescriptors;

public class ReadFromStorage {
  public static Pipeline createPipeline(Options options) {
    var pipeline = Pipeline.create(options);
    pipeline
        // Read from a text file.
        .apply(TextIO.read().from(
            "gs://" + options.getBucket() + "/*.txt"))
        .apply(
            MapElements.into(TypeDescriptors.strings())
                .via(
                    (x -> {
                      System.out.println(x);
                      return x;
                    })));
    return pipeline;
  }
}

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

  • קוראים את מאמרי העזרה של TextIO API.
  • כאן אפשר לעיין ברשימת התבניות ש-Google סיפקה.