/*需求：
在指定文件目录下的所有文件中，检索某一特定字符串所出现的行，将这些行的内容输出到本地文件系统的输出文件夹中。这个程序
假定只有第一层目录下的文件才有效，而且，假定文件都是文本文件。为了防止单个的输出文件过大，这里还加了一个文件最大行数限制
当文件行数达到最大值时，便关闭此文件，创建另外的文件继续保存。保存的结果文件名为1，2,3,4，..,*/
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.File;
import java.io.IOException;
import java.util.Scanner;
public class ResultFilter {
//实现逻辑：获取改目录下所有文件的信息，对每一个文件（打开，循环读取数据，写入目标位置，关闭文件，最后关闭输出文件）
public static void main(String[] args) throws IOException {
Configuration conf = new Configuration();
//hdfs和local分别对应HDFS实例和本地文件系统实例
FileSystem hdfs = FileSystem.get(conf);
FileSystem local = FileSystem.get(conf);
Path inputDir, localFile;
int singleFileLines, numLines, numFiles;
FileStatus[] inputFiles;
FSDataInputStream in = null;
FSDataOutputStream out = null;
Scanner scan;
String str;
byte[] buf;
if (args.length != 4) {
//输入参数数量不够，提示参数格式后终止程序执行
System.err.println("usage resultFilter <dfs path><local path>" + "<math str><single file lines>");
return;
}
inputDir = new Path(args[0]);//获取输入的hdfs路径
singleFileLines = Integer.parseInt(args[3]);//获取结果文件的最大行数
inputFiles = hdfs.listStatus(inputDir);//获取目录信息
numLines = 0;//文件的行数从0开始
numFiles = 1;//输出文件从1开始编号
localFile = new Path(args[1]); //获取输出的本地文件路径
if (local.exists(localFile)) //若目标路径存在，则删除
local.delete(localFile, true);
for (int j = 0; j < inputFiles.length; j++) {
if (inputFiles[j].isDir() == true)//忽略子目录
continue;
System.out.println(inputFiles[j].getPath().getName());
in = hdfs.open(inputFiles[j].getPath());
scan = new Scanner(in);
while (scan.hasNext()) {//循环读取文本数据
str = scan.nextLine();
if (str.indexOf(args[2]) == -1)
continue;//如果该行没有match字符串，则忽略之
numLines++;
if (numLines == 1)//如果是1，说明需要新建文件了
{
localFile = new Path(args[1] + File.separator + numFiles);
out = local.create(localFile);//创建文件
numFiles++;
}
buf = (str + "\n").getBytes();
out.write(buf, 0, buf.length);//将字符串写入输出流
if (numLines == singleFileLines)//如果已满足相应行数，关闭文件
{
out.close();
numLines = 0;//把行数变为0，重新开始计数
}
}// end of while
scan.close();
in.close();
}// end of for
if (out != null)
out.close();
}// end of maain
} //end of resultFilter