hadoop - 如何使用 AVRO org.apache.avro.mapreduce 接口(interface)进行编程？

Question

我所有的程序都是用 hadoop 的新 MR1 接口(interface) (org.apache.hadoop.mapreduce) 编写的，所以我也想使用 avro 的新 org.apache.avro.mapreduce。但这对我不起作用。

该程序接受 avro 数据的输入并输出相同的数据。我的程序背后的主要思想是根据 avro 包装的键/值对 hadoop 的 Mapper 和 Reducer 进行子类化。这是我的工作驱动程序的一部分:

    AvroJob.setInputKeySchema(job, NetflowRecord.getClassSchema());
    AvroJob.setOutputKeySchema(job, NetflowRecord.getClassSchema());

    job.setMapperClass(MyAvroMap.class);
    job.setReducerClass(MyAvroReduce.class);

    job.setInputFormatClass(AvroKeyInputFormat.class);
    job.setOutputFormatClass(AvroKeyOutputFormat.class);

    job.setMapOutputKeyClass(AvroKey.class);
    job.setMapOutputValueClass(AvroValue.class);

    job.setOutputKeyClass(AvroKey.class);
    job.setOutputValueClass(NullWritable.class);

MyAvroMap 和 MyAvroReduce 子类的定义分别是

public static class MyAvroMap extends Mapper<AvroKey<NetflowRecord>, NullWritable,
            AvroKey<CharSequence>, AvroValue<NetflowRecord>>{ ... }

public static class MyAvroReduce extends Reducer<AvroKey<CharSequence>, AvroValue<NetflowRecord>, 
                AvroKey<NetflowRecord>, NullWritable>{ ... }

methioned NetflowRecord 是我的 avro 记录类。我遇到了运行异常

java.lang.ClassCastException: class org.apache.avro.hadoop.io.AvroKey

通过阅读hadoop和avro的源代码，我发现这个异常是由JobConf抛出的，以确保map key是WritableComparable的子类，像这样(hadoop1.2.1, line759)

WritableComparator.get(getMapOutputKeyClass().asSubclass(WritableComparable.class));

但 avro 显示 AvroKey 和 AvroValue 只是一个简单的包装器没有子类化 hadoop 的 Writable* 接口(interface)。

我相信，即使不进行测试，我也可以使用旧的 mapred 接口(interface)，但这不是我想要的。你能给我一些关于使用纯 org.apache.avro.mapreduce 接口(interface)编程的例子或解释吗？？

此致

贾敏

Answer 1

经过苦苦寻找，借助这个补丁https://issues.apache.org/jira/browse/AVRO-593 ,我发现每个 AvroKey 和 AvroValue 包装器都必须在作业配置中有一个架构定义。这就是我错过的。

这里我有两个选择:

1. 如果保持 MyAvroMap 和 MyAvroReduce 不变，我必须为 CharSequence 定义一个架构，并使用 AvroJob 声明此架构以用于 Mapper 输出，例如

   AvroJob.setMapOutputKeySchema(工作，<“为字符序列定义的模式”>)；AvroJob.setMapOutputValueSchema(作业, NetflowRecord.getClassSchema());

2. 通过将 Mapper 输出键/值更改为 Text/AvroValue，我只需要为 Mapper 输出值添加模式声明，例如

   job.setMapOutputKeyClass(Text.class);AvroJob.setMapOutputValueSchema(作业, NetflowRecord.getClassSchema());

使用 mapreduce API，我们不再需要子类化 AvroMapper 和 AvroReducer。在这里，我在我的代码中实现了选项 2 而没有附加模式定义。

贾敏