如何在两个map reduce作业之间传递变量答案

【问题标题】：How to pass variable between two map reduce jobs如何在两个map reduce作业之间传递变量
【发布时间】：2012-12-01 00:04:55
【问题描述】：

我已经链接了两个 Map reduce 作业。 Job1 将只有一个减速器，我正在计算一个浮点值。我想在 Job2 的减速器中使用这个值。这是我的主要方法设置。

public static String GlobalVriable;
public static void main(String[] args) throws Exception {

        int runs = 0;
        for (; runs < 10; runs++) {
            String inputPath = "part-r-000" + nf.format(runs);
            String outputPath = "part-r-000" + nf.format(runs + 1);
            MyProgram.MR1(inputPath);
            MyProgram.MR2(inputPath, outputPath);
        }
    }

    public static void MR1(String inputPath)
            throws IOException, InterruptedException, ClassNotFoundException {

        Configuration conf = new Configuration();
        conf.set("var1","");
        Job job = new Job(conf, "This is job1");
        job.setJarByClass(MyProgram.class);
        job.setMapperClass(MyMapper1.class);
        job.setReducerClass(MyReduce1.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FloatWritable.class);
        FileInputFormat.addInputPath(job, new Path(inputPath));
        job.waitForCompletion(true);
        GlobalVriable = conf.get("var1"); // I am getting NULL here
    }

    public static void MR2(String inputPath, String outputPath)
            throws IOException, InterruptedException, ClassNotFoundException {

        Configuration conf = new Configuration();
        Job job = new Job(conf, "This is job2");
        ...
    }

    public static class MyReduce1 extends
        Reducer<Text, FloatWritable, Text, FloatWritable> {

    public void reduce(Text key, Iterable<FloatWritable> values, Context context)
            throws IOException, InterruptedException {

        float s = 0;
        for (FloatWritable val : values) {
            s += val.get();
        }

        String sum = Float.toString(s);
        context.getConfiguration().set("var1", sum);
    }
}

如您所见，我需要多次迭代整个程序。我的 Job1 正在从输入中计算一个数字。由于它只是一个数字和很多迭代，我不想将它写入 HDFS 并从中读取。有没有办法共享在 Myreducer1 中计算的值并在 Myreducer2 中使用它。

更新：我尝试使用 conf.set 和 conf.get 传递值。该值没有被传递。

【问题讨论】：

reducer 正在生成 text:text 输出以及单个浮点值？
哦，我的错，它是一个 FloatWritable。我会编辑它。
仍然有两种数据从 reducer 中输出？
reducer 的键是文本，值是 FloatWritable。我将遍历 FloatWritable 并计算总和。我不打算将其写入文件，我只是想将这笔款项传递给下一份工作

标签： hadoop mapreduce hdfs

【解决方案1】：

这是通过计数器传回浮点值的方法...

首先，在第一个 reducer 中，通过乘以 1000（例如，保持 3 位精度）将浮点值转换为 long 并将结果放入计数器：

public void cleanup(Context context) {

    long result = (long) (floatValue * 1000);
    context.getCounter("Result","Result").increment(result); 

}

在驱动类中，检索长值并将其转换回浮点数：

public static void MR1(String inputPath)
        throws IOException, InterruptedException, ClassNotFoundException {

    Configuration conf = new Configuration();
    Job job = new Job(conf, "This is job1");
    job.setJarByClass(MyProgram.class);
    job.setMapperClass(MyMapper1.class);
    job.setReducerClass(MyReduce1.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(FloatWritable.class);
    FileInputFormat.addInputPath(job, new Path(inputPath));
    job.waitForCompletion(true);

    long result = job.getCounters().findCounter("Result","Result").getValue();
    float value = ((float)result) / 1000;

}

【讨论】：

非常感谢！！虽然我应该将它写入文件并从中读取，但这会增加我的运行时间。我也想知道，因为我没有将 job1 的输出写入文件，是否可以避免创建输出目录。当我设置 FileOutputFormat.setOutputPath 时，Hadoop 会创建一个。

【解决方案2】：

您可以为此使用ZooKeeper。它非常适合像这样的任何工作间协调或消息传递。

【讨论】：

你能详细说明一下吗？

【解决方案3】：

你不能把MR1的返回类型改为int（或任何合适的数据类型）并返回你计算出来的数字吗：

    int myNumber = MyProgram.MR1(inputPath);

然后将参数添加到MR2 并使用您计算的数字调用它：

    MyProgram.MR2(inputPath, outputPath, myNumber);

【讨论】：

您的回答让我有些困惑。返回我在 reducer 中计算的数字是什么意思？我虽然减速器将值写入 HDFS 而不返回它们？你能解释一下如果有超过 2 个减速器会发生什么吗？
reducer 不输出单个数字，它输出键/值对。你正在计算的数字是多少？你是怎么计算的？
我已经更新了我的代码。我尝试使用 conf 传递，但变量中的值始终为 null