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

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

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

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

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

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

אפשר גם להשתמש בדוגמאות הקוד מהמדריך או מהמדריך לתחילת העבודה עם 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, שמופעל כברירת מחדל כשמשתמשים בRunner v2 (נדרש). אלגוריתם ההתאמה האוטומטית לעומס ב-Dataflow לוקח בחשבון את הפיגור המשוער של סנכרון שינויים בזרמי נתונים, שאפשר לעקוב אחריו בדף Dataflow monitoring (מעקב אחרי Dataflow) בקטע Backlog. משתמשים בדגל --maxNumWorkers כשפורסים משימה כדי להגביל את מספר העובדים.

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

מגבלות

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

‫Dataflow Runner V2

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

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

Snapshots

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

כפילויות

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

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

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

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

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

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

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

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

  2. אם אתם משתמשים בספק זהויות חיצוני (IdP), קודם אתם צריכים להיכנס ל-CLI של gcloud באמצעות המאגר המאוחד לניהול זהויות.

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

    gcloud auth application-default login

    אם אתם משתמשים ב-Cloud Shell, אין צורך לבצע את הפעולה הזו.

    אם מוחזרת שגיאת אימות ואתם משתמשים בספק זהויות חיצוני (IdP), ודאו ש נכנסתם ל-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, אתם צריכים את התפקידים הבאים:

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

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

הוספת מחבר 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, מוסבר איך ליצור צינור עיבוד נתונים ולהשתמש בו כדי להמיר צמדי מפתח/ערך למחרוזת. כל צמד מורכב ממפתח שורה ומאובייקט 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:

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

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

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

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

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

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

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