משימת Hadoop MapReduce עם Bigtable

בדוגמה הזו נעשה שימוש ב-Hadoop כדי לבצע משימת מיפוי וצמצום פשוטה שסופרת את מספר הפעמים שמילה מופיעה בקובץ טקסט. העבודה MapReduce משתמשת ב-Bigtable כדי לאחסן את התוצאות של פעולת המיפוי. הקוד של הדוגמה הזו נמצא במאגר GitHub‏ GoogleCloudPlatform/cloud-bigtable-examples, בספרייה java/dataproc-wordcount.

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

כדי להשתמש בדוגמאות של 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 באמצעות המאגר המאוחד לניהול זהויות.

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

סקירה כללית של דוגמת הקוד

דוגמת הקוד מספקת ממשק שורת פקודה (CLI) פשוט שמקבל כקלט קובץ טקסט אחד או יותר ושם של טבלה, מוצא את כל המילים שמופיעות בקובץ וסופר כמה פעמים כל מילה מופיעה. הלוגיקה של MapReduce מופיעה במחלקת WordCountHBase.

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

public static class TokenizerMapper extends
    Mapper<Object, Text, ImmutableBytesWritable, IntWritable> {

  private final static IntWritable one = new IntWritable(1);

  @Override
  public void map(Object key, Text value, Context context) throws IOException,
      InterruptedException {
    StringTokenizer itr = new StringTokenizer(value.toString());
    ImmutableBytesWritable word = new ImmutableBytesWritable();
    while (itr.hasMoreTokens()) {
      word.set(Bytes.toBytes(itr.nextToken()));
      context.write(word, one);
    }
  }
}

לאחר מכן, פונקציית reducer מסכמת את הערכים של כל מפתח וכותבת את התוצאות לטבלת Bigtable שציינתם. כל מפתח שורה הוא מילה מקובץ הטקסט. כל שורה מכילה עמודה cf:count, שמכילה את מספר הפעמים שמפתח השורה מופיע בקובץ הטקסט.

public static class MyTableReducer extends
    TableReducer<ImmutableBytesWritable, IntWritable, ImmutableBytesWritable> {

  @Override
  public void reduce(ImmutableBytesWritable key, Iterable<IntWritable> values, Context context)
      throws IOException, InterruptedException {
    int sum = sum(values);
    Put put = new Put(key.get());
    put.addColumn(COLUMN_FAMILY, COUNT_COLUMN_NAME, Bytes.toBytes(sum));
    context.write(null, put);
  }

  public int sum(Iterable<IntWritable> values) {
    int i = 0;
    for (IntWritable val : values) {
      i += val.get();
    }
    return i;
  }
}