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

wlwcrazw  于 2021-06-03  发布在  Hadoop
关注(0)|答案(3)|浏览(326)

我把两份Map缩小工作都锁上了。job1将只有一个减速机,我正在计算一个浮点值。我想在job2的reducer中使用这个值。这是我的主要方法设置。

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);
    }
}

如您所见,我需要多次迭代整个程序。我的工作是从输入中计算一个数字。因为它只是一个单一的数字和大量的迭代,我不想把它写入hdfs并从中读取。有没有一种方法可以共享myreducer1中计算的值并在myreducer2中使用它。
更新:我尝试使用conf.set&conf.get传递值。未传递值。

x4shl7ld

x4shl7ld1#

你就不能改变退货类型吗 MR1int (或任何适当的数据类型)并返回您计算的数字:

int myNumber = MyProgram.MR1(inputPath);

然后将参数添加到 MR2 并用你的计算号码呼叫它:

MyProgram.MR2(inputPath, outputPath, myNumber);
gdrx4gfi

gdrx4gfi2#

下面是如何通过计数器传回浮点值。。。
首先,在第一个减速机中,将浮点值乘以1000(例如,保持3位精度)并将结果放入计数器,从而将浮点值转换为long:

public void cleanup(Context context) {

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

}

在driver类中,检索long值并将其转换回float:

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;

}
i34xakig

i34xakig3#

你可以用zookeeper来做这个。这是伟大的任何工作间的协调或信息传递这样。

相关问题