משימת Hadoop MapReduce עם Bigtable

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

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

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

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

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

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

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

   gcloud auth application-default login

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

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

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

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

קודם, פונקציית mapper מבצעת טוקניזציה של התוכן בקובץ הטקסט ויוצרת זוגות של מפתח-ערך, כאשר המפתח הוא מילה מקובץ הטקסט והערך הוא 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;
  }
}