Ví dụ MapReduce

Question

Tôi đã đọc về mapreduce và tôi đã tự hỏi về một kịch bản cụ thể. Giả sử chúng ta có một vài tệp (ví dụ fileA, fileB, fileC), mỗi tệp chứa nhiều số nguyên. Nếu chúng tôi muốn sắp xếp các con số từ tất cả các tệp để tạo một cái gì đó như thế này:

23 fileA 
34 fileB 
35 fileA 
60 fileA 
60 fileC 

quy trình bản đồ và giảm bớt hoạt động như thế nào?

Hiện tại, đây là những gì tôi có nhưng nó không hoàn toàn chính xác;

1. (fileName, fileContent) -> (map to) (Number, fileName)

2. loại chìa khóa, cặp giá trị tạm thời và nhận được (Number, (list of){fileName1, fileName2...})

3. Giảm cặp tạm thời và nhận được

   (Number, fileName1) 
   (Number, fileName2) 
   

   và vân vân

Vấn đề là trong giai đoạn sắp xếp, tên tệp có thể không theo thứ tự bảng chữ cái và do đó phần giảm sẽ không tạo ra kết quả chính xác. Ai đó có thể cung cấp một số cái nhìn sâu sắc về cách tiếp cận chính xác cho kịch bản này?

Answer 1

Cách tốt nhất để đạt được điều này là thông qua phân loại phụ. Bạn cần phải sắp xếp cả hai khóa (trong số trường hợp của bạn) và các giá trị (trong tên tập tin trường hợp của bạn). Trong Hadoop, đầu ra của trình ánh xạ chỉ được sắp xếp trên các khóa.

Điều này có thể đạt được bằng cách sử dụng phím tổng hợp: khóa là kết hợp cả số và tên tệp. Ví dụ: đối với bản ghi đầu tiên, khóa sẽ là (23, fileA), thay vì chỉ (23).

Bạn có thể đọc về loại thứ ở đây: https://www.safaribooksonline.com/library/view/data-algorithms/9781491906170/ch01.html

Bạn cũng có thể đi qua phần "Secondary Sắp xếp", trong "Hadoop The Definitive Hướng dẫn" cuốn sách.

Vì mục đích đơn giản, tôi đã viết một chương trình để đạt được điều tương tự.

Trong chương trình này, các khóa được sắp xếp theo mặc định bởi những người lập bản đồ. Tôi đã viết một logic để sắp xếp các giá trị ở phía giảm tốc. Vì vậy, nó sẽ chăm sóc phân loại cả khóa và giá trị và tạo ra kết quả mong muốn đó.

Tiếp theo là chương trình:

package com.myorg.hadooptests; 

import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.*; 
import org.apache.hadoop.mapreduce.Job; 
import org.apache.hadoop.mapreduce.Mapper; 
import org.apache.hadoop.mapreduce.Reducer; 
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 

import java.io.IOException; 
import java.util.*; 

public class SortedValue { 


    public static class SortedValueMapper 
      extends Mapper<LongWritable, Text , Text, IntWritable>{ 

     public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 

      String[] tokens = value.toString().split(" "); 

      if(tokens.length == 2) { 
       context.write(new Text(tokens[1]), new IntWritable(Integer.parseInt(tokens[0]))); 
      } 
     } 
    } 

    public static class SortedValueReducer 
      extends Reducer<Text, IntWritable, IntWritable, Text> { 

     Map<String, ArrayList<Integer>> valueMap = new HashMap<String, ArrayList<Integer>>(); 

     public void reduce(Text key, Iterable<IntWritable> values, 
          Context context) throws IOException, InterruptedException { 

      String keyStr = key.toString(); 
      ArrayList<Integer> storedValues = valueMap.get(keyStr); 

      for (IntWritable value : values) { 
       if (storedValues == null) { 
        storedValues = new ArrayList<Integer>(); 
        valueMap.put(keyStr, storedValues); 
       } 
       storedValues.add(value.get()); 
      } 

      Collections.sort(storedValues); 
      for (Integer val : storedValues) { 
       context.write(new IntWritable(val), key); 
      } 
     } 
    } 

    public static void main(String[] args) throws Exception { 

     Configuration conf = new Configuration(); 

     Job job = Job.getInstance(conf, "CompositeKeyExample"); 
     job.setJarByClass(SortedValue.class); 
     job.setMapperClass(SortedValueMapper.class); 
     job.setReducerClass(SortedValueReducer.class); 

     job.setMapOutputKeyClass(Text.class); 
     job.setMapOutputValueClass(IntWritable.class); 

     job.setOutputKeyClass(IntWritable.class); 
     job.setOutputValueClass(Text.class); 

     FileInputFormat.addInputPath(job, new Path("/in/in1.txt")); 
     FileOutputFormat.setOutputPath(job, new Path("/out/")); 

     System.exit(job.waitForCompletion(true) ? 0:1); 

    } 
} 

Mapper Logic:

1. phân tích mỗi dòng. Giả định rằng khóa và giá trị được phân cách bằng ký tự trống ("").
2. Nếu dòng chứa 2 thẻ, nó phát ra (tên tệp, giá trị số nguyên). Ví dụ: cho bản ghi đầu tiên, nó phát ra (fileA, 23).

Giảm Logic:

1. Nó đặt (key, value) cặp trong một HashMap, nơi chính là tên tập tin và giá trị là một danh sách các số nguyên cho tập tin đó. Ví dụ: đối với tệpA, các giá trị được lưu trữ sẽ là 23, 34 và 35

2. Cuối cùng, nó sắp xếp các giá trị cho một khóa cụ thể và cho mỗi giá trị phát ra (giá trị, khóa) từ bộ giảm tốc. Ví dụ: cho fileA, các hồ sơ đầu ra là: (23, fileA), (34 tuổi, fileA) và (35, fileA)

tôi chạy chương trình này cho các đầu vào sau:

34 fileB 
35 fileA 
60 fileC 
60 fileA 
23 fileA 

tôi đầu ra sau:

23  fileA 
35  fileA 
60  fileA 
34  fileB 
60  fileC