【问题标题】:Is it possible to associate an instance of an object with one file while it's being mapped by a map-only mapred Job?是否可以将一个对象的实例与一个文件相关联,同时它被一个仅映射的映射作业映射?
【发布时间】:2013-09-25 20:43:13
【问题描述】:

我想使用一个 HashSet,它在映射一个文件时存在/对它起作用,然后在映射下一个文件时重置/重新创建。我已修改 TextInputFormat 以覆盖 isSplitable 以返回 false,这样文件就不会被拆分,而是由 Mappers 整体处理。有可能做这样的事情吗?还是有其他方法可以减少对 Accumulo 表的写入?

让我从我不相信我想要一个全局变量开始。我只是想确保唯一性,从而减少向我的 Accumulo 表写入的突变。

我的项目是将分片示例中的 Index.java 文件的功能从线性 accumulo 客户端程序转换为使用 mapreduce 功能的程序,同时仍然在 Accumulo 中创建相同的表。它需要 mapreduce,因为这是流行语,本质上它比线性程序运行 TB 级数据的速度更快。

这里是索引代码供参考: http://grepcode.com/file/repo1.maven.org/maven2/org.apache.accumulo/examples-simple/1.4.0/org/apache/accumulo/examples/simple/shard/Index.java

该程序使用 BatchWriter 将 Mutations 写入 Accumulo,并基于每个文件执行此操作。为了确保它不会写入不必要的突变并确保唯一性(尽管我确实相信 Accumulo 最终会通过压缩合并相同的键),Index.java 有一个 HashSet 用于确定之前是否已经运行过一个单词。这一切都比较容易理解。

转移到仅地图的 mapreduce 作业更复杂。

这是我的映射尝试,从我看到的 Accumulo 表的部分输出来看,这似乎有点工作,但与线性程序 Index.java 相比,它的运行速度真的很慢

public static class MapClass extends Mapper<LongWritable,Text,Text,Mutation> {
        private HashSet<String> tokensSeen = new HashSet<String>();
        @Override
        public void map(LongWritable key, Text value, Context output) throws IOException {
            FileSplit fileSplit = (FileSplit)output.getInputSplit();
            System.out.println("FilePath " + fileSplit.getPath().toString());
            String filePath = fileSplit.getPath().toString();
            filePath = filePath.replace("unprocessed", "processed");

            String[] words = value.toString().split("\\W+");

            for (String word : words) {
                Mutation mutation = new Mutation(genPartition(filePath.hashCode() % 10));
                word = word.toLowerCase();
                if(!tokensSeen.contains(word)) {
                    tokensSeen.add(word);
                    mutation.put(new Text(word), new Text(filePath), new Value(new byte[0]));
                }

                try {
                    output.write(null, mutation);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

缓慢的问题可能是我在一个测试实例上运行所有这些,这是一个带有 ZooKeeper 和 Accumulo 的单节点 Hadoop 实例。 如果是这样的话,我只需要找到唯一性的解决方案。

非常感谢您提供的任何帮助或建议。

【问题讨论】:

    标签: java hadoop mapreduce accumulo


    【解决方案1】:

    Mapper 具有 setupcleanup 方法,您可以重写它们以更干净地处理此类事情。 setup 被调用一次,然后map 被调用多次(每条记录一次),然后cleanup 在最后被调用一次。想法是您在 setup 方法中创建 HashSet,在 map 中构建它,并在 cleanup 中提交所有内容,或者在必要时定期刷新对 map 的一些调用。

    但是,在迁移到真正的集群之前,您几乎肯定不会看到运行时的任何改进。与简单的线性程序相比,单节点测试实例几乎没有任何优势,只是在获得真正的 hadoop 集群后,相同的代码会运行得更快。

    【讨论】:

      猜你喜欢
      • 1970-01-01
      • 1970-01-01
      • 2022-12-07
      • 1970-01-01
      • 1970-01-01
      • 2012-10-16
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      相关资源
      最近更新 更多