如何让每个 hadoop 映射器获取文件对，即整个输入文件 (.csv) 和整个元数据文件 (.json)答案

【问题标题】：How to make each hadoop mapper to get a file pair i.e. a whole input file (.csv) and a whole meta data file (.json)如何让每个 hadoop 映射器获取文件对，即整个输入文件 (.csv) 和整个元数据文件 (.json)
【发布时间】：2015-02-08 07:46:36
【问题描述】：

我在同一个文件夹中有成百上千个输入文件 (.csv) 和元数据文件 (.json)。 $HDFS_ROOT/输入文件夹

// 输入数据.csv文件

input_1.csv, input_2.csv..input_N.csv

// 输入元数据.json文件

input_1.json, input_2.json..input_N.json

谁能告诉我如何让每个映射器获取文件对，即整个输入文件 (.csv) 及其元数据文件 (.json)。

注意：input_i.csv 和 input_i.json 应该转到同一个映射器，以便输入及其元数据都有意义进行验证。

我尝试了什么：我尝试分别使用从 FileInputFormat 和 RecordReader 扩展的 WholeFileInputFormat 和 WholeFileRecordReader。这仅适用于 .csv 文件。此外，我将 .json 文件放入分布式缓存中，以便映射器访问。这不是一个好的解决方案。

【问题讨论】：

你尝试了什么？这不是其他人为您编码的编码服务。
冷静。我尝试分别使用从 FileInputFormat 和 RecordReader 扩展的 WholeFileInputFormat 和 WholeFileRecordReader。这仅适用于 .csv 文件。此外，我将 .json 文件放入分布式缓存中。这不是一个好的解决方案。
为什么不把这个留给reducer呢？使用 MultipleInputs 将您的 CSV 和 JSON 数据导入不同的映射器，然后在 reduce 阶段加入它们。这将是处理这个问题的预期方式。我认为这对你来说不是很好，因为你想做进一步的加入，所以也许这个工作在 MapReduce 之外处理会更好？级联可以让你做 map -> reduce -> reduce。
感谢您的意见。刚刚在这里也看到了 MultipleInputs 的示例：lichun.cc/blog/2012/05/hadoop-multipleinputs-usage 如果我理解您基于 MR 的想法，请纠正我。每个 Mapper 都会将键作为“input_i”发出，并将 .csv 和 .json 的相应（整个）数据作为文本发出，在 reducer 中，我们将获得键“input_i”和 iterable 的值。现在，reducer 可以进行验证、转换等，或者只是将键“input_i”和值作为 .json 数据 + .csv 数据发出。 MR 的下一阶段现在可以访问 .json + .csv 数据进行处理。
我看到的一些问题是：reducer 的文本值顺序可能不同，即第一个值可能是由 .csv 数据或 .json 数据组成的文本。此外，如果 .csv 的文件大小大于 64MB 或 128MB 怎么办。那么值可能 > 2。我们如何确定 iterable 中的值是 .csv 数据还是 .json 数据。

标签： java hadoop mapreduce hbase

【解决方案1】：

在不使用昂贵的 Reducer 的情况下解决这个问题的关键是 InputSplits。每个 InputFormat 都有方法 getSplits，单个拆分是单个 Mapper 的输入，有多少个 Mapper，就有多少个 InputSplits。在映射器中，可以访问 InputSplit 的实例：

@Override
protected void setup(Context context) throws IOException, InterruptedException {
     System.out.println("TRACE 1 " + context.getConfiguration().getClass().getName());
     System.out.println("TRACE 2 " + context.getTaskAttemptID().toString());
     System.out.println("TRACE 3 " + context.getInputSplit().toString());

}

基于此，我过去使用过 3 种方法：

1) context.getInputSplit() 返回一个FileSplit 的实例，它具有Path getPath() 方法。但是您必须注意可能会环绕 FileSplit 的 CombineFileSplit 和 TaggedInputSplit。使用 CombineFileSplit 如果您不覆盖 CombineFileInputFormat.pools 周围的默认行为，那么您可能会在同一 Mapper 中混合具有不同结构的记录而无法区分它们；

2) 更简单的方法是使用 context.getInputSplit().toString()，返回的字符串将包含 InputSplit 附加到的路径，效果很好使用 MultipleInputs，但不使用 CombineFileInputFormat。它有点脏，因为您受 toString() 方法的支配，不建议将其用于生产系统，但对于快速原型来说已经足够了；

3) 要定义自己的代理 InputFormat 和 InputSplit 实现，类似于 MultipleInputs 方法使用的，它依赖于 DelegatingInputFormat包裹可以读取数据的 InputFormat 的 InputSplit，但将它们放在 TaggedInputSplit 内，请参阅源代码。在您的情况下，您可以在自己的 InputFormat 和 InputSplits 中隐藏元数据逻辑，并使 Mappers 不知道如何将文件与元数据匹配。您也可以直接将输入路径关联到元数据，而无需依赖命名约定。这种方法非常适合生产系统。

【讨论】：