java - 从 Java 代码运行 Linux Hadoop fs 命令

Question

我正在尝试从两个合并到文件的 java 代码运行命令!命令是:

hadoop fs -cat /user/clouder/Index_1/part-r-00000 /user/cloudera/Index_2/part-r-00000 | hadoop fs -put - /user/cloudera/mergedfile

该命令在 Cloudera 终端上运行完美，但是当我从 java 代码运行相同的命令时，它在控制台上显示合并的内容，但不会在 HDFS 上的指定路径中创建合并文件。如果合并文件已经存在，那么它输出文件的早期数据而不是新合并的数据，如果文件不存在，那么它不会创建新文件。如果上面的命令在终端上运行会创建新文件，如果不存在则给出文件存在的错误。

我的java代码如下:

process p;

try{

        p =Runtime.getRuntime().exec("hadoop fs -cat /user/cloudera/Index_1/part-r-00000 /user/cloudera/Index_2/part-r-00000 | hadoop fs -put - /user/cloudera/mergedfile");
        BufferredReader br=new BufferedReader(new InputStreamReader(p.getInputStream()));

        while(s=br.readLine())!=null)
        {
            System.out.println(s);
        }
    }

catch(Exception e)
    {
        System.out.println(e.getMessage());
    }

我的目的是在存在文件的情况下进行替换，或者在 Java 代码不存在的情况下创建一个新文件。

Answer 1

要使用 Java 运行 HDFS 命令，您应该使用 HDFS Java API .这是一个 code sample from javased.com如何使用它来合并这些文件:

/** 
 * @param inputFiles a glob expression of the files to be merged
 * @param outputFile a destination file path
 * @param deleteSource delete source files after merging
 * @return
 * @throws IOException
 */
private static Path mergeTextFiles(String inputFiles,String outputFile,boolean deleteSource,boolean deleteDestinationFileIfExist) throws IOException {
  JobConf conf=new JobConf(FileMerger.class);
  FileSystem fs=FileSystem.get(conf);
  Path inputPath=new Path(inputFiles);
  Path outputPath=new Path(outputFile);
  if (deleteDestinationFileIfExist) {
    if (fs.exists(outputPath)) {
      fs.delete(outputPath,false);
      sLogger.info("Warning: remove destination file since it already exists...");
    }
  }
 else {
    Preconditions.checkArgument(!fs.exists(outputPath),new IOException("Destination file already exists..."));
  }
  FileUtil.copyMerge(fs,inputPath,fs,outputPath,deleteSource,conf,FILE_CONTENT_DELIMITER);
  sLogger.info("Successfully merge " + inputPath.toString() + " to "+ outputFile);
  return outputPath;
}

在这种情况下，您需要事先使用 FileUtil class 将要合并的文件复制到 1 个目录中.稍后您将采用这样的目录路径并将其作为 inputFiles 参数传递。:

JobConf conf=new JobConf(FileMerger.class);
FileSystem fs=FileSystem.get(conf);
String tmpDir = "/user/cloudera/tmp_dir";
Path[] paths = {new Path("/user/clouder/Index_1/part-r-00000"), new Path("/user/clouder/Index_2/part-r-00000")};
Path pathToInputs = FileUtil.copy(fs, paths, fs, new Path(tmpDir));
mergeTextFiles(tmpDir, "/user/cloudera/mergedfile", false, true);