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

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

הוספת תלות בספרייה של Google Cloud Platform

כדי להשתמש במחבר 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

Go 

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, אפשר להפעיל את מחבר ה-I/O של 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 | עמודה אחת תפוקה (בייטים) תפוקה (אלמנטים)
קריאה ‫320MBps ‫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;
  }
}

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