הזרמת שינויים באמצעות Dataflow

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

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

חלופות ליצירת צינור עיבוד נתונים משלכם

אם אתם לא רוצים ליצור צינור Dataflow משלכם, אתם יכולים להשתמש באחת מהאפשרויות הבאות.

אפשר להשתמש בתבנית Dataflow שסופקה על ידי Google.

• סנכרון שינויים בזרמי נתונים מ-Bigtable ל-BigQuery
• סנכרון שינויים בזרמי נתונים מ-Bigtable ל-Pub/Sub

אפשר גם להשתמש בדוגמאות הקוד מהמדריך או מהמדריך לתחילת העבודה עם Bigtable כנקודת התחלה לקוד.

• יצירת טבלה עם הפעלה של זרם שינויים ותיעוד שינויים

• עיבוד של שינויים ב-Bigtable

מוודאים שהקוד שאתם יוצרים משתמש ב-google cloud libraries-bom גרסה 26.14.0 ואילך.

פרטי המחבר

השיטה של מחבר Bigtable Beam, ‏ BigtableIO.readChangeStream, מאפשרת לקרוא זרם של רשומות שינוי נתונים (ChangeStreamMutation) שאפשר לעבד. מחבר Bigtable Beam הוא רכיב של מאגר Apache Beam GitHub. תיאור של קוד המחבר מופיע בתגובות בכתובת BigtableIO.java.

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

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

קיבוץ נתונים ללא זמני אירועים

רשומות של שינויי נתונים שמוזרמות באמצעות מחבר Bigtable Beam לא תואמות לפונקציות Dataflow שתלויות בזמני אירועים.

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

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

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

התאמה אוטומטית לעומס (Automatic scaling)

המחבר תומך בשינוי גודל אוטומטי ב-Dataflow, שמופעל כברירת מחדל כשמשתמשים בPortable Runner (נדרש). אלגוריתם ההתאמה לעומס ב-Dataflow לוקח בחשבון את הפיגור המשוער של זרם השינויים, שאפשר לעקוב אחריו בדף Dataflow monitoring (מעקב אחרי Dataflow) בקטע Backlog. משתמשים בדגל --maxNumWorkers כשפורסים משימה כדי להגביל את מספר העובדים.

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

מגבלות

לפני שמשתמשים במחבר Bigtable Beam עם Dataflow, חשוב לשים לב להגבלות הבאות.

‫Dataflow Portable Runner

אפשר להפעיל את המחבר רק באמצעות Dataflow Portable Runner. כדי להפעיל את האפשרות הזו, מציינים --experiments=use_runner_v2 בארגומנטים של שורת הפקודה. הפעלה באמצעות Streaming Java Runner גורמת לצינור להיכשל עם החריגה הבאה:

java.lang.UnsupportedOperationException: BundleFinalizer unsupported by non-portable Dataflow

תמונות מצב

המחבר לא תומך בתמונות מצב של Dataflow.

כפילויות

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

הפעלה מחדש של פייפליין

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

לפני שמתחילים

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

מגדירים אימות

כדי להשתמש בדוגמאות של Java שבדף הזה בסביבת פיתוח מקומית, מתקינים ומפעילים את ה-CLI של gcloud, ואז מגדירים את Application Default Credentials באמצעות פרטי הכניסה של המשתמש.

1. התקינו את ה-CLI של Google Cloud.

2. הגדירו שה-CLI של gcloud ישתמש בזהות המאוחדת שלכם.

   איך נכנסים ל-CLI של gcloud באמצעות הזהות המאוחדת?

3. יוצרים פרטי כניסה לאימות מקומי עבור חשבון המשתמש:

   gcloud auth application-default login

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

מידע נוסף זמין במאמר הגדרת אימות לסביבת פיתוח מקומית.

למידע נוסף על הגדרה של אימות בסביבת ייצור, ראו הגדרה של Application Default Credentials לקוד שפועל ב- Google Cloud .

הפעלת שינוי השידור החי

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

שינוי טבלת המטא-נתונים של השידור

כשמבצעים סטרימינג של שינויים באמצעות Dataflow, מחבר Bigtable Beam יוצר טבלת מטא-נתונים בשם __change_stream_md_table כברירת מחדל. בטבלת המטא-נתונים של זרם השינויים מנוהל המצב התפעולי של המחבר, ונשמרים בה מטא-נתונים לגבי רשומות של שינויים בנתונים.

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

מידע נוסף על סטרימינג של שינויים מ-Bigtable באמצעות מחבר Bigtable Beam זמין במסמכי התיעוד של BigtableIO.

התפקידים הנדרשים

כדי לקבל את ההרשאות שדרושות לקריאת נתוני שינויים ב-Bigtable באמצעות Dataflow, צריך לבקש מהאדמין להקצות לכם את תפקידי ה-IAM הבאים.

כדי לקרוא את השינויים מ-Bigtable, צריך את התפקיד הבא:

• Bigtable Administrator (roles/bigtable.admin) במופע Bigtable שמכיל את הטבלה שמתכננים להזרים ממנה שינויים

כדי להריץ את משימת Dataflow, אתם צריכים את התפקידים הבאים:

• מפתח Dataflow (roles/dataflow.developer) בפרויקט שמכיל את משאבי Cloud
• ‫Dataflow Worker (roles/dataflow.worker) בפרויקט שמכיל את משאבי Cloud
• התפקיד Storage Object Admin‏ (roles/storage.objectAdmin) בקטגוריות של Cloud Storage שבהן אתם מתכננים להשתמש

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

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

הוספת מחבר Bigtable Beam כתלות

מוסיפים קוד דומה לתלות הבאה לקובץ Maven pom.xml. הגרסה צריכה להיות 2.48.0 ואילך.

<dependencies>
  <dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
    <version>VERSION</version>
  </dependency>
</dependencies>

קריאת השינוי בשידור החי

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

בדוגמאות הקוד בקטע הזה, שנכתבו ב-Java, מוסבר איך ליצור צינור (pipeline) ולהשתמש בו כדי להמיר צמדי מפתח/ערך למחרוזת. כל צמד מורכב ממפתח שורה ומאובייקט ChangeStreamMutation. הצינור ממיר את הרשומות של כל אובייקט למחרוזת מופרדת בפסיקים.

פיתוח הפייפליין

בדוגמת הקוד הבאה ב-Java אפשר לראות איך יוצרים את צינור העיבוד:

BigtableOptions options =
    PipelineOptionsFactory.fromArgs(args).withValidation().as(BigtableOptions.class);
Pipeline p = Pipeline.create(options);

final Instant startTime = Instant.now();

p.apply(
        "Read Change Stream",
        BigtableIO.readChangeStream()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId())
            .withAppProfileId(options.getBigtableAppProfile())
            .withStartTime(startTime))
    .apply(
        "Flatten Mutation Entries",
        FlatMapElements.into(TypeDescriptors.strings())
            .via(ChangeStreamsHelloWorld::mutationEntriesToString))
    .apply(
        "Print mutations",
        ParDo.of(
            new DoFn<String, Void>() { // a DoFn as an anonymous inner class instance
              @ProcessElement
              public void processElement(@Element String mutation) {
                System.out.println("Change captured: " + mutation);
              }
            }));
p.run();

עיבוד הרשומות של שינויי הנתונים

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

רשימה של סוגי הרשומות שרשומה של שינוי נתונים יכולה להכיל מופיעה במאמר מה כוללת רשומה של שינוי נתונים.

static List<String> mutationEntriesToString(KV<ByteString, ChangeStreamMutation> mutationPair) {
  List<String> mutations = new ArrayList<>();
  String rowKey = mutationPair.getKey().toStringUtf8();
  ChangeStreamMutation mutation = mutationPair.getValue();
  MutationType mutationType = mutation.getType();
  for (Entry entry : mutation.getEntries()) {
    if (entry instanceof SetCell) {
      mutations.add(setCellToString(rowKey, mutationType, (SetCell) entry));
    } else if (entry instanceof DeleteCells) {
      mutations.add(deleteCellsToString(rowKey, mutationType, (DeleteCells) entry));
    } else if (entry instanceof DeleteFamily) {
      // Note: DeleteRow mutations are mapped into one DeleteFamily per-family
      mutations.add(deleteFamilyToString(rowKey, mutationType, (DeleteFamily) entry));
    } else {
      throw new RuntimeException("Entry type not supported.");
    }
  }
  return mutations;
}

בדוגמה הזו, רשומה של write מומרת:

private static String setCellToString(String rowKey, MutationType mutationType, SetCell setCell) {
  List<String> mutationParts =
      Arrays.asList(
          rowKey,
          mutationType.name(),
          "SetCell",
          setCell.getFamilyName(),
          setCell.getQualifier().toStringUtf8(),
          setCell.getValue().toStringUtf8());
  return String.join(",", mutationParts);
}

בדוגמה הזו, ערך של מחיקת תאים מומר:

private static String deleteCellsToString(
    String rowKey, MutationType mutationType, DeleteCells deleteCells) {
  String timestampRange =
      deleteCells.getTimestampRange().getStart() + "-" + deleteCells.getTimestampRange().getEnd();
  List<String> mutationParts =
      Arrays.asList(
          rowKey,
          mutationType.name(),
          "DeleteCells",
          deleteCells.getFamilyName(),
          deleteCells.getQualifier().toStringUtf8(),
          timestampRange);
  return String.join(",", mutationParts);
}

בדוגמה הזו, רשומה של מחיקת קבוצת עמודות מומרת:

private static String deleteFamilyToString(
    String rowKey, MutationType mutationType, DeleteFamily deleteFamily) {
  List<String> mutationParts =
      Arrays.asList(rowKey, mutationType.name(), "DeleteFamily", deleteFamily.getFamilyName());
  return String.join(",", mutationParts);
}

מעקב

המשאבים הבאים במסוף Google Cloud מאפשרים לכם לעקוב אחרי משאביGoogle Cloud בזמן שאתם מריצים פייפליין של Dataflow כדי לקרוא שינויים בסטרימינג ב-Bigtable:

• מעקב ב-Dataflow
• תובנות לגבי מערכת Bigtable

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

• בדף התובנות של מערכת Bigtable, בודקים את המדדים הבאים:
  • נתונים של CPU utilization by change streams במדד סנכרון שינויים בזרמי נתוניםcpu_load_by_app_profile_by_method_by_table. התרשים מציג את ההשפעה של זרם השינויים על השימוש במעבד (CPU) של האשכול.
  • שינוי השימוש באחסון של הסטרימינג (בייט) (change_stream_log_used_bytes).

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

  • אם מדד רעננות הנתונים גבוה באופן עקבי מהסף, יכול להיות שאין מספיק משאבים בצינור. מומלץ להוסיף עוד עובדי Dataflow.
  • אם יש מספיק workers ב-Dataflow אבל עדכניות הנתונים עולה או שהיא גבוהה באופן עקבי, צריך לפנות אל Google Cloud התמיכה.

• מדד processing_delay_from_commit_timestamp_MEANDataflow יכול להראות לכם את זמן העיבוד הממוצע של רשומות שינוי נתונים במהלך משך החיים של העבודה.

המדד server/latencies של Bigtable לא שימושי כשעוקבים אחרי צינור Dataflow שקורא זרם שינויים של Bigtable, כי הוא משקף את משך בקשת הסטרימינג ולא את זמן האחזור של עיבוד רשומת שינוי הנתונים. חביון גבוה בזרם שינויים לא אומר שהבקשות מעובדות לאט, אלא שהחיבור היה פתוח למשך הזמן הזה.

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

• איך כותבים מ-Dataflow ל-Cloud Storage
• רשימה מלאה של מדדי המעקב שזמינים ב-Bigtable
• שימוש במעקב כדי לבדוק את מדדי Dataflow.