使用Hadoop查找包含特定字符串的文件

Question

我有大約1000 files，每個文件的大小爲1GB。我需要在所有這些1000 files中找到一個字符串，以及哪些文件包含該特定的字符串。我正在使用Hadoop文件系統，所有這些1000 files都在Hadoop文件系統中。

所有的1000 files都在真實文件夾下，所以如果我這樣做，我會得到所有的1000 files。我需要找到哪些文件包含特定字符串hello在實際文件夾下。

bash-3.00$ hadoop fs -ls /technology/dps/real 

這是hdfs-

row format delimited 
fields terminated by '\29' 
collection items terminated by ',' 
map keys terminated by ':' 
stored as textfile 

我的數據結構，我怎麼能寫MapReduce作業做這方面的問題，這樣我可以找到哪些文件包含一個特定的字符串？任何簡單的例子都會對我有很大的幫助。

更新： -

隨着Unix下使用grep的，我可以解決上述問題的情況下，但它是非常非常緩慢，需要花費很多的時候得到的實際輸出 -

hadoop fs -ls /technology/dps/real | awk '{print $8}' | while read f; do hadoop fs -cat $f | grep cec7051a1380a47a4497a107fecb84c1 >/dev/null && echo $f; done 

所以這就是我一直在尋找一些MapReduce作業做這種問題的原因...

Answer 1

這聽起來像你正在尋找一個類似grep的程序，這是很容易使用Hadoop Streaming實現（Hadoop的Java API的將工作太）：

首先，編寫輸出的名稱的映射正在處理的行包含您的搜索字符串時正在處理的文件。我用Python，但任何語言將工作：

#!/usr/bin/env python 
import os 
import sys 

SEARCH_STRING = os.environ["SEARCH_STRING"] 

for line in sys.stdin: 
    if SEARCH_STRING in line.split(): 
     print os.environ["map_input_file"] 

此代碼讀取SEARCH_STRING環境變量的搜索字符串。在這裏，我分割輸入行並檢查搜索字符串是否匹配任何分割;你可以改變它來執行子字符串搜索或使用正則表達式來檢查匹配。

接下來，使用這種映射，與沒有減速運行Hadoop的數據流作業：

$ bin/hadoop jar contrib/streaming/hadoop-streaming-*.jar \ 
    -D mapred.reduce.tasks=0 
    -input hdfs:///data \ 
    -mapper search.py \ 
    -file search.py \ 
    -output /search_results \ 
    -cmdenv SEARCH_STRING="Apache" 

輸出將被寫在幾個部分;獲得匹配的列表，你可以簡單的貓文件（只要不是太大）：

$ bin/hadoop fs -cat /search_results/part-* 
hdfs://localhost/data/CHANGES.txt 
hdfs://localhost/data/CHANGES.txt 
hdfs://localhost/data/ivy.xml 
hdfs://localhost/data/README.txt 
... 

Answer 2

你可以嘗試這樣的事情，雖然我不知道這是否是一種有效的方式去了解它。讓我知道它是否有效 - 我沒有測試它或任何東西。

你可以使用它是這樣的：的Java SearchFiles /技術/ DPS /實際你好確保你當然從相應的目錄中運行它。

import java.io.BufferedReader; 
import java.io.FileReader; 
import java.io.File; 
import java.io.IOException; 
import java.util.ArrayList; 
import java.util.Scanner; 

public class SearchFiles { 

    public static void main(String[] args) throws IOException { 
     if (args.length < 2) { 
      System.err.println("Usage: [search-dir] [search-string]"); 
      return; 
     } 
     File searchDir = new File(args[0]); 
     String searchString = args[1]; 
     ArrayList<File> matches = checkFiles(searchDir.listFiles(), searchString, new ArrayList<File>()); 
     System.out.println("These files contain '" + searchString + "':"); 
     for (File file : matches) { 
      System.out.println(file.getPath()); 
     } 
    } 

    private static ArrayList<File> checkFiles(File[] files, String search, ArrayList<File> acc) throws IOException { 
     for (File file : files) { 
      if (file.isDirectory()) { 
       checkFiles(file.listFiles(), search, acc); 
      } else { 
       if (fileContainsString(file, search)) { 
        acc.add(file); 
       } 
      } 
     } 
     return acc; 
    } 

    private static boolean fileContainsString(File file, String search) throws IOException { 
     BufferedReader in = new BufferedReader(new FileReader(file)); 
     String line; 
     while ((line = in.readLine()) != null) { 
      if (line.contains(search)) { 
       in.close(); 
       return true; 
      } 
     } 
     in.close(); 
     return false; 
    } 
} 

Answer 3

要獲得當前正在處理的文件名，這樣做：

((FileSplit) context.getInputSplit()).getPath().getName() 

當你被記錄搜索你的文件記錄，當你看到hello，發出上述路徑（也許行或還要別的嗎）。

將reducer的數量設置爲0，他們在這裏沒有做任何事情。

---

'行格式分隔'是否意味着行被換行符分隔？在這種情況下，TextInputFormat和LineRecordReader可以在此處正常工作。