将对象从驱动程序传递给MapReduce

这与边数据分配问题有关.

侧数据分发有两种方法.

1)分布式缓存

2)配置

当您拥有要共享的对象时,我们可以使用Configuration类.

此讨论将取决于Configuration类,以使整个群集中的Object可用,所有Mappers和(或)Reducers都可以访问.这里的方法很简单.使用Configuration类的setString(String,String)setter来实现此任务.必须共享的Object在驱动程序端序列化为java字符串,并反序列化为Mapper或Reducer处的对象.

在下面的示例代码中,我使用了com.google.gson.Gson类来进行简单的序列化和反序列化.您也可以使用Java序列化.

表示您需要共享的对象的类

 public class TestBean {
    String string1;
    String string2;
    public TestBean(String test1, String test2) {
        super();
        this.string1 = test1;
        this.string2 = test2;
    }
    public TestBean() {
        this("", "");
    }
    public String getString1() {
        return string1;
    }
    public void setString1(String test1) {
        this.string1 = test1;
    }
    public String getString2() {
        return string2;
    }
    public void setString2(String test2) {
        this.string2 = test2;
    }
}

您可以在其中设置配置的主类

public class GSONTestDriver {
    public static void main(String[] args) throws Exception {
        System.out.println("In Main");
        Configuration conf = new Configuration();
        TestBean testB1 = new TestBean("Hello1","Gson1");
        TestBean testB2 = new TestBean("Hello2","Gson2");
        Gson gson = new Gson();
        String testSerialization1 = gson.toJson(testB1);
        String testSerialization2 = gson.toJson(testB2);
        conf.set("instance1", testSerialization1);
        conf.set("instance2", testSerialization2);
        Job job = new Job(conf, " GSON Test");
        job.setJarByClass(GSONTestDriver.class);
        job.setMapperClass(GSONTestMapper.class);
        job.setNumReduceTasks(0);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.waitForCompletion(true);
    }
}

mapper类,您可以从中检索对象

public class GSONTestMapper extends
Mapper<LongWritable, Text, Text, NullWritable> {
    Configuration conf;
    String inst1;
    String inst2;
    public void setup(Context context) {
        conf = context.getConfiguration();
        inst1 = conf.get("instance1");
        inst2 = conf.get("instance2");
        Gson gson = new Gson();
        TestBean tb1 = gson.fromJson(inst1, TestBean.class);
        System.out.println(tb1.getString1());
        System.out.println(tb1.getString2());
        TestBean tb2 = gson.fromJson(inst2, TestBean.class);
        System.out.println(tb2.getString1());
        System.out.println(tb2.getString2());
    } 
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        context.write(value,NullWritable.get());
    }
}

使用com.google.gson.Gson类的toJson(Object src)方法将bean转换为序列化的Json String.然后,序列化的Json字符串作为值通过配置实例传递,并通过Mapper中的名称访问.使用相同Gson类的fromJson(String json,Class classOfT)方法对字符串进行反序列化.您可以放置​​对象而不是我的测试bean.