Cần trợ giúp triển khai thuật toán này với bản đồ Hadoop MapReduce

Question

tôi có thuật toán sẽ đi qua một tập dữ liệu lớn đọc một số tệp văn bản và tìm kiếm các cụm từ cụ thể trong các dòng đó. Tôi có nó thực hiện trong Java, nhưng tôi didnt muốn đăng mã để nó không nhìn tôi đang tìm kiếm một ai đó để thực hiện nó cho tôi, nhưng nó là sự thật tôi thực sự cần rất nhiều sự giúp đỡ !!! Điều này đã không được lên kế hoạch cho dự án của tôi, nhưng tập dữ liệu hóa ra là rất lớn, vì vậy giáo viên nói với tôi rằng tôi phải làm như thế này.

EDIT (tôi đã không làm rõ tôi previos version) Tập hợp dữ liệu tôi có là trên một cụm Hadoop, và tôi nên thực hiện MapReduce nó

Tôi đã đọc về MapReduce và thaught rằng tôi lần đầu tiên thực hiện thực hiện tiêu chuẩn và sau đó nó sẽ được nhiều hơn/ít dễ dàng hơn để làm điều đó với mapreduce. Nhưng đã không xảy ra, kể từ thuật toán là khá ngu ngốc và không có gì đặc biệt, và bản đồ giảm ... tôi không thể quấn tâm trí của tôi xung quanh nó.

Vì vậy, đây là một thời gian ngắn pseudo code của thuật toán của tôi

LIST termList (there is method that creates this list from lucene index) 
FOLDER topFolder 

INPUT topFolder 
IF it is folder and not empty 
    list files (there are 30 sub folders inside) 
    FOR EACH sub folder 
     GET file "CheckedFile.txt" 
     analyze(CheckedFile) 
    ENDFOR 
END IF 


Method ANALYZE(CheckedFile) 

read CheckedFile 
WHILE CheckedFile has next line 
    GET line 
    FOR(loops through termList) 
      GET third word from line 
      IF third word = term from list 
     append whole line to string buffer 
    ENDIF 
ENDFOR 
END WHILE 
OUTPUT string buffer to file 

Ngoài ra, như bạn có thể thấy, mỗi lần khi "phân tích" được gọi, tập tin mới đã được tạo ra, tôi hiểu bản đồ mà giảm là khó khăn để ghi vào nhiều kết quả đầu ra ???

Tôi hiểu bản đồ trực quan, và ví dụ của tôi có vẻ hoàn toàn phù hợp với mapreduce, nhưng khi nói đến điều này, rõ ràng là tôi không biết đủ và tôi đang STUCK!

Vui lòng trợ giúp.

Answer 1

Bạn chỉ có thể sử dụng trình giảm tốc trống và phân vùng công việc của mình để chạy một trình ánh xạ đơn cho mỗi tệp. Mỗi người lập bản đồ sẽ tạo tệp đầu ra của riêng nó trong thư mục đầu ra của bạn.

Answer 2

Bản đồ Giảm dễ dàng được triển khai bằng một số tính năng đồng thời Java 6 đẹp, đặc biệt là Tương lai, Có thể gọi và ExecutorService.

Tôi tạo ra một Callable rằng sẽ phân tích một tập tin theo cách mà bạn chỉ định

public class FileAnalyser implements Callable<String> { 

    private Scanner scanner; 
    private List<String> termList; 

    public FileAnalyser(String filename, List<String> termList) throws FileNotFoundException { 
    this.termList = termList; 
    scanner = new Scanner(new File(filename)); 
    } 

    @Override 
    public String call() throws Exception { 
    StringBuilder buffer = new StringBuilder(); 
    while (scanner.hasNextLine()) { 
     String line = scanner.nextLine(); 
     String[] tokens = line.split(" "); 
     if ((tokens.length >= 3) && (inTermList(tokens[2]))) 
     buffer.append(line); 
    } 
    return buffer.toString(); 
    } 

    private boolean inTermList(String term) { 
    return termList.contains(term); 
    } 
} 

Chúng ta cần tạo một callable mới cho mỗi tập tin được tìm thấy và nộp này với dịch vụ thi hành di chúc. Kết quả của việc gửi là một tương lai mà chúng ta có thể sử dụng sau này để có được kết quả của phân tích cú pháp tệp.

public class Analayser { 

    private static final int THREAD_COUNT = 10; 

    public static void main(String[] args) { 

    //All callables will be submitted to this executor service 
    //Play around with THREAD_COUNT for optimum performance 
    ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT); 

    //Store all futures in this list so we can refer to them easily 
    List<Future<String>> futureList = new ArrayList<Future<String>>(); 

    //Some random term list, I don't know what you're using. 
    List<String> termList = new ArrayList<String>(); 
    termList.add("terma"); 
    termList.add("termb"); 

    //For each file you find, create a new FileAnalyser callable and submit 
    //this to the executor service. Add the future to the list 
    //so we can check back on the result later 
    for each filename in all files { 
     try { 
     Callable<String> worker = new FileAnalyser(filename, termList); 
     Future<String> future = executor.submit(worker); 
     futureList.add(future); 
     } 
     catch (FileNotFoundException fnfe) { 
     //If the file doesn't exist at this point we can probably ignore, 
     //but I'll leave that for you to decide. 
     System.err.println("Unable to create future for " + filename); 
     fnfe.printStackTrace(System.err); 
     } 
    } 

    //You may want to wait at this point, until all threads have finished 
    //You could maybe loop through each future until allDone() holds true 
    //for each of them. 

    //Loop over all finished futures and do something with the result 
    //from each 
    for (Future<String> current : futureList) { 
     String result = current.get(); 
     //Do something with the result from this future 
    } 
    } 
} 

Ví dụ của tôi ở đây không đầy đủ và hiệu quả. Tôi đã không được coi là cỡ mẫu, nếu nó thực sự khổng lồ bạn có thể giữ looping qua futureList, loại bỏ các yếu tố đó đã kết thúc, một cái gì đó tương tự như:

while (futureList.size() > 0) { 
     for (Future<String> current : futureList) { 
     if (current.isDone()) { 
      String result = current.get(); 
      //Do something with result 
      futureList.remove(current); 
      break; //We have modified the list during iteration, best break out of for-loop 
     } 
     } 
} 

Hoặc bạn có thể thực hiện một thiết lập kiểu sản xuất-tiêu dùng nơi nhà sản xuất gửi các cuộc gọi đến dịch vụ thi hành và tạo ra một tương lai và người tiêu dùng lấy kết quả của tương lai và loại bỏ sau này trong tương lai.

Điều này có thể yêu cầu sản phẩm và người tiêu dùng phải là chủ đề của chính họ và danh sách được đồng bộ hóa để thêm/xóa tương lai.

Mọi câu hỏi, vui lòng hỏi.