Hadoop MultipleOutPutFormat 和连接查询

Question

我正在处理一个 hadoop 任务，该任务之前在目录中填充了几个文件说

部分-o第1部分第二部分

我根据要求修改了此任务，并使用 MultipleOutputs 捕获更多输出。所以现在目录结构看起来像

第0部分第1部分第2部分输出-1输出2输出-3

问题:之前很少有作业使用此目录进行映射侧外部连接，但现在该作业必须仅采用部分 -* 文件进行连接并丢弃其余文件。

我尝试将输入作为“,”分隔的目录即 /part-1,/part-2,*/part-3

并根据以下表达式jobConf.set("mapred.join.expr", CompositeInputFormat.compose(outer, KeyValueTextInputFormat.class, path[]))

现在我的路径 [] 包含 5 个整体，以前是 3 个，在某种程度上，起始三个索引具有 /part-1、/part-2、*/part-3 路径和像之前一样休息两个。

我不确定我这样做是否正确，请建议我应该怎么做才能使此连接像以前一样在没有输出-* 文件的情况下工作。

使用上述方法抛出以下异常。

java.io.IOException:子 1 的不一致拆分基数 (12/6) org.apache.hadoop.mapred.join.Parser$CNode.getSplits(Parser.java:369) org.apache.hadoop.mapred.join.CompositeInputFormat.getSplits(CompositeInputFormat.java:117) org.apache.hadoop.mapred.JobClient.writeOldSplits(JobClient.java:810) org.apache.hadoop.mapred.JobClient.submitJobInternal(JobClient.java:781) org.apache.hadoop.mapred.JobClient.submitJob(JobClient.java:730)

JobConf 值如下。

 jobConf.setMapperClass(MyMapper.class);
    jobConf.setReducerClass(MyReducer.class);

  String[] foldersToJoin = StringUtils.split(getInputString(), Constants.COMMA);
    Path[] pathsToJoin =  new Path[foldersToJoin.length];
    int i = 0;
    for(String folder : foldersToJoin){
        pathsToJoin[i++] = new Path(folder);
    }

 FileOutputFormat.setOutputPath(jobConf, new Path("/MyOutPUT"));

    jobConf.setInputFormat(CompositeInputFormat.class);
    jobConf.set("mapred.join.expr", CompositeInputFormat.compose(Constants.OUTER_JOIN_OP,
            KeyValueTextInputFormat.class, pathsToJoin));

    jobConf.setOutputFormat(TextOutputFormat.class);
    jobConf.setOutputKeyClass(Text.class);
    jobConf.setOutputValueClass(Text.class);

    MultipleOutputs.addNamedOutput(jobConf, CHANGE_SET_A,
            TextOutputFormat.class, Text.class, Text.class);
    MultipleOutputs.addNamedOutput(jobConf, CHANGE_SET_B,
            TextOutputFormat.class, Text.class, Text.class);
    MultipleOutputs.addNamedOutput(jobConf,CHANGE_SET_C,
            TextOutputFormat.class, Text.class, Text.class);
    MultipleOutputs.addNamedOutput(jobConf, CHANGE_SET_D,
            TextOutputFormat.class, Text.class, Text.class);

我将文件夹中的所有文件附加为“，”分隔的字符串，后面是代码

  mapred.join.expr = CompositeInputFormat.compose(Constants.OUTER_JOIN_OP,
        KeyValueTextInputFormat.class, pathsToJoin)

   where pathsToJoin =  new Path[]{new Path["/home/hadoop/folder1/part-1"],
  new Path["/home/hadoop/folder1/part-2"],new Path["/home/hadoop/folder1/part-3"],
   new Path["/home/hadoop/folder2"],new Path["/home/hadoop/folder3"] }

所以基本上我只是尝试将 folder1 中存在的 part-* 文件与 folder2 和 folder3 合并

任何文档或指向此类场景的任何链接都会有很大帮助。

Answer 1

好的，我想我现在明白了。

你已经执行了一些将它们的内容输出到一个文件夹的作业:

job1 -> folder1
job2 -> folder2
job3 -> folder3

现在您想使用 CompositeInputFormat 合并每个文件夹中每个 part-r-x 的输出，并在单个映射器中处理

map0 - merged contents of folder1/part-r-0, folder2/part-r-0, folder3/part-r-0
map1 - merged contents of folder1/part-r-1, folder2/part-r-1, folder3/part-r-1
.. and so on ..

增加的复杂性是一个或多个作业使用了 MultipleOutputs，因此您拥有的不是文件夹 1 中的 part-r-x 文件

job1 -> folder1/part-x and folder1/output-x

当您开始使用 CompositeInputFormat 时，它会出错，因为 folder1 的文件比 folder2 和 3 多

在这种情况下，我认为您需要修改 mapred.join.expr 值以使用一些 glob:

// use glob for folder1, to only include the part-x files (ignoring the output-x files)
CompositeInputFormat.compose(Constants.OUTER_JOIN_OP, KeyValueTextInputFormat.class,
   new Path[] {
       new Path('folder1/part-*'),
       new Path('folder2/part-r-*'),
       new Path('folder3/part-r-*'),
   });