עיבוד של שינוי בשידור חי ב-Bigtable

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

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

המדריך הזה מיועד למשתמשים טכניים שמכירים את כתיבת הקוד ואת פריסת צינורות עיבוד הנתונים ב- Cloud de Confiance by S3NS.

מטרות

במדריך הזה מוסבר איך:

  • יוצרים טבלת Bigtable עם הפעלת זרם שינויים.
  • פריסת צינור ב-Dataflow שמבצע טרנספורמציה של נתוני הסטרים של השינויים ומייצא אותם.
  • צפייה בתוצאות של צינור עיבוד הנתונים.

עלויות

במסמך הזה משתמשים ברכיבים הבאים של Cloud de Confiance by S3NS, והשימוש בהם כרוך בתשלום:

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

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

    מתקינים את ה-CLI של Google Cloud ואז נכנסים ל-CLI של gcloud באמצעות הזהות המאוחדת שלכם. אחרי שנכנסתם לחשבון, אתחלו את ה-CLI של Google Cloud באמצעות הפקודה הבאה: 

    gcloud init

    יוצרים או בוחרים Cloud de Confiance פרויקט.

    תפקידים שנדרשים כדי לבחור או ליצור פרויקט

    • Select a project: כדי לבחור פרויקט לא צריך תפקיד IAM ספציפי – אפשר לבחור כל פרויקט שקיבלתם בו תפקיד.
    • יצירת פרויקט: כדי ליצור פרויקט, צריך את התפקיד Project Creator (roles/resourcemanager.projectCreator), שכולל את ההרשאה resourcemanager.projects.create. איך מקצים תפקידים

    • יוצרים Cloud de Confiance פרויקט:

      gcloud projects create PROJECT_ID

      מחליפים את PROJECT_ID בשם של פרויקט Cloud de Confiance שיוצרים.

    • בוחרים את הפרויקט שיצרתם: Cloud de Confiance 

      gcloud config set project PROJECT_ID

      מחליפים את PROJECT_ID בשם הפרויקט ב- Cloud de Confiance .

    מוודאים שהחיוב מופעל בפרויקט Cloud de Confiance .

    מפעילים את ממשקי ה-API של Dataflow,‏ Cloud Bigtable API,‏ Cloud Bigtable Admin API ו-Cloud Storage, אם הם עדיין לא מופעלים:

    תפקידים שנדרשים להפעלת ממשקי API

    כדי להפעיל ממשקי API, צריך את ההרשאה serviceusage.services.enable. אם יצרתם את הפרויקט, סביר להניח שכבר יש לכם את ההרשאה הזו דרך התפקיד 'בעלים' (roles/owner). אחרת, תוכלו לקבל את ההרשאה הזו דרך התפקיד 'אדמין בממשק Service Usage' (roles/serviceusage.serviceUsageAdmin). איך מקצים תפקידים 

    gcloud services enable dataflow.googleapis.com bigtable.googleapis.com bigtableadmin.googleapis.com storage.googleapis.com
  1. מעדכנים ומתקינים את cbt CLI . 
    gcloud components update
gcloud components install cbt

הכנת הסביבה

קבל את הקוד

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

git clone https://github.com/GoogleCloudPlatform/java-docs-samples.git
cd java-docs-samples/bigtable/beam/change-streams

יצירת קטגוריה

  • יוצרים קטגוריה של Cloud Storage:
    ‫
    gcloud storage buckets create gs://BUCKET_NAME
     מחליפים את BUCKET_NAME בשם קטגוריה שעומד בקריטריונים לשמות של קטגוריות.

    • יצירת מכונה של Bigtable

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

    צור טבלה

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

    יוצרים את הטבלה.

    gcloud bigtable instances tables create song-rank \
--column-families=cf --change-stream-retention-period=7d \
--instance=BIGTABLE_INSTANCE_ID --project=PROJECT_ID

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

    • ‫PROJECT_ID: מזהה הפרויקט שבו אתם משתמשים
    • ‫BIGTABLE_INSTANCE_ID: המזהה של המכונה שמכילה את הטבלה החדשה

    הפעלת הפייפליין

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

    1. קריאת השינוי בשידור החי
    2. קבלת שם השיר
    3. מקבץ את אירועי ההאזנה לשירים בחלונות של N שניות
    4. סופר את חמשת השירים המובילים
    5. הפלט של התוצאות

    מריצים את הפייפליין.

    mvn compile exec:java -Dexec.mainClass=SongRank \
"-Dexec.args=--project=PROJECT_ID --bigtableProjectId=PROJECT_ID \
--bigtableInstanceId=BIGTABLE_INSTANCE_ID --bigtableTableId=song-rank \
--outputLocation=gs://BUCKET_NAME/ \
--runner=dataflow --region=BIGTABLE_REGION --experiments=use_runner_v2"

    מחליפים את BIGTABLE_REGION במזהה האזור שבו נמצא מופע Bigtable, לדוגמה, us-east5.

    הסבר על צינור עיבוד הנתונים

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

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

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

    p.apply(
        "Stream from Bigtable",
        BigtableIO.readChangeStream()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId())
            .withAppProfileId(options.getBigtableAppProfile())

    )

    איך מוצאים את שם השיר

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

    private static class ExtractSongName extends DoFn<KV<ByteString, ChangeStreamMutation>, String> {

  @DoFn.ProcessElement
  public void processElement(ProcessContext c) {

    for (Entry e : Objects.requireNonNull(Objects.requireNonNull(c.element()).getValue())
        .getEntries()) {
      if (e instanceof SetCell) {
        SetCell setCell = (SetCell) e;
        if ("cf".equals(setCell.getFamilyName())
            && "song".equals(setCell.getQualifier().toStringUtf8())) {
          c.output(setCell.getValue().toStringUtf8());
        }
      }
    }
  }
}

    ספירת חמשת השירים המובילים

    אפשר להשתמש בפונקציות המובנות של Beam‏ Count ו-Top.of כדי לקבל את חמשת השירים המובילים בחלון הנוכחי.

    .apply(Count.perElement())
.apply("Top songs", Top.of(5, new SongComparator()).withoutDefaults())

    הצגת התוצאות

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

    .apply("Print", ParDo.of(new PrintFn()))
.apply(
    "Collect at least 10 elements or 1 minute of elements",
    Window.<String>into(new GlobalWindows())
        .triggering(
            Repeatedly.forever(
                AfterFirst.of(
                    AfterPane.elementCountAtLeast(10),
                    AfterProcessingTime
                        .pastFirstElementInPane()
                        .plusDelayOf(Duration.standardMinutes(1)
                        )
                )
            ))
        .discardingFiredPanes())
.apply(
    "Output top songs",
    TextIO.write()
        .to(options.getOutputLocation() + "song-charts/")
        .withSuffix(".txt")
        .withNumShards(1)
        .withWindowedWrites()
);

    הצגת הפייפליין

    1. נכנסים לדף Dataflow במסוף Cloud de Confiance .

      מעבר אל Dataflow

    2. לוחצים על העבודה עם שם שמתחיל ב-song-rank.

    3. בתחתית המסך, לוחצים על הצגה כדי לפתוח את חלונית היומנים.

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

    כתיבה בשידור

    משתמשים ב-cbtCLI כדי לכתוב מספר של האזנות לשירים עבור משתמשים שונים בטבלה song-rank. התכונה הזו נועדה לכתוב נתונים במשך כמה דקות כדי לדמות האזנות לשירים שמוזרמים לאורך זמן.

    cbt -instance=BIGTABLE_INSTANCE_ID -project=PROJECT_ID import \
song-rank song-rank-data.csv  column-family=cf batch-size=1

    צפייה בפלט

    קוראים את הפלט ב-Cloud Storage כדי לראות את השירים הכי פופולריים.

    gcloud storage cat gs://BUCKET_NAME/song-charts/GlobalWindow-pane-0-00000-of-00001.txt

    פלט לדוגמה:

    2023-07-06T19:53:38.232Z [KV{The Wheels on the Bus, 199}, KV{Twinkle, Twinkle, Little Star, 199}, KV{Ode to Joy , 192}, KV{Row, Row, Row Your Boat, 186}, KV{Take Me Out to the Ball Game, 182}]
2023-07-06T19:53:49.536Z [KV{Old MacDonald Had a Farm, 20}, KV{Take Me Out to the Ball Game, 18}, KV{Für Elise, 17}, KV{Ode to Joy , 15}, KV{Mary Had a Little Lamb, 12}]
2023-07-06T19:53:50.425Z [KV{Twinkle, Twinkle, Little Star, 20}, KV{The Wheels on the Bus, 17}, KV{Row, Row, Row Your Boat, 13}, KV{Happy Birthday to You, 12}, KV{Over the Rainbow, 9}]

    הסרת המשאבים

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

    מחיקת הפרויקט

      כדי למחוק Cloud de Confiance פרויקט:

      gcloud projects delete PROJECT_ID

    מחיקת משאבים בודדים

    1. מוחקים את הקטגוריה והקבצים.

      gcloud storage rm --recursive gs://BUCKET_NAME/

    2. משביתים את זרם השינויים בטבלה.

      gcloud bigtable instances tables update song-rank --instance=BIGTABLE_INSTANCE_ID \
--clear-change-stream-retention-period

    3. מוחקים את הטבלה song-rank.

      cbt -instance=BIGTABLE_INSTANCE_ID -project=PROJECT_ID deletetable song-rank

    4. עוצרים את צינור עיבוד הנתונים של שינוי הנתונים.

      1. מציגים את רשימת המשרות כדי לקבל את מזהה המשרה.

        gcloud dataflow jobs list --region=BIGTABLE_REGION

      2. ביטול המשימה.

        gcloud dataflow jobs cancel JOB_ID --region=BIGTABLE_REGION

        מחליפים את JOB_ID במזהה המשימה שמוצג אחרי הפקודה הקודמת.

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