- android - RelativeLayout 背景可绘制重叠内容
- android - 如何链接 cpufeatures lib 以获取 native android 库?
- java - OnItemClickListener 不起作用,但 OnLongItemClickListener 在自定义 ListView 中起作用
- java - Android 文件转字符串
我有一个 mapreduce 程序,它从 avro 数据中读取数据,对其进行处理并输出 avro 数据。我有这个 avro 数据的模式,假设有 4 列。我使用 GenericData.Record 来写入 avro 数据。
现在,我使用具有 5 列的模式在此数据之上创建一个 pig 关系。第 5 列是新的,具有 avsc 文件中定义的默认值。根据我的理解,我应该能够使用带有一列的新模式读取旧数据(由 4 列生成)。相反,我收到一条错误消息 - Trying to access non-exist column.
我错过了什么?
Mapreduce 驱动程序代码
Job job = Job.getInstance(getConf());
job.setJarByClass(DeltaCaptureMRJobDriverWithSameSchema.class);
job.setJobName("CDC");
job.getConfiguration().setBoolean("mapreduce.input.fileinputformat.input.dir.recursive", true);
//This is required to use avro-1.7.6 and above
job.getConfiguration().set("mapreduce.job.user.classpath.first", "true");
FileInputFormat.addInputPaths(job, args[0]);
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.setInputFormatClass(AvroKeyInputFormat.class);
job.setMapperClass(DeltaCaptureMapperMultiPaths.class);
Schema schema = new Schema.Parser().parse(new File(args[2]));
AvroJob.setInputKeySchema(job, schema);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(AvroValue.class);
AvroJob.setMapOutputValueSchema(job, schema);
job.setOutputFormatClass(AvroKeyOutputFormat.class);
job.setReducerClass(DeltaCaptureReducerMultiPaths.class);
AvroJob.setOutputKeySchema(job, schema);
job.setOutputKeyClass(AvroKey.class);
return (job.waitForCompletion(true) ? 0 : 1);
映射器代码
public class DeltaCaptureMapperMultiPaths extends Mapper<AvroKey<GenericData.Record>, NullWritable, Text , AvroValue<GenericData.Record>> {
private static final Logger LOG = Logger.getLogger(DeltaCaptureMapperMultiPaths.class);
@Override
public void map(AvroKey<GenericData.Record> key, NullWritable value, Context context) throws IOException, InterruptedException {
try {
System.out.println("Specific Record - " + key);
System.out.println("Datum :: " + key.datum());
System.out.println("Schema :: " + key.datum().getSchema());
AvroValue<GenericData.Record> outValue = new AvroValue<GenericData.Record>(key.datum());
System.out.println("Generic Record (out) - " + key.datum());
context.write(new Text(key.datum().get("id") +""), outValue);
} catch (Exception e) {
e.printStackTrace();
}
}
}
Reducer 代码
public class DeltaCaptureReducerMultiPaths extends Reducer<Text, AvroValue<GenericData.Record>, AvroKey<GenericData.Record>, NullWritable> {
@Override
public void reduce(Text key, Iterable<AvroValue<GenericData.Record>> values, Context context) throws IOException, InterruptedException {
for(AvroValue<GenericData.Record> value : values) {
AvroKey<GenericData.Record> outKey = new AvroKey<GenericData.Record>(value.datum());
context.write(outKey, NullWritable.get());
}
}
}
假设 MR 输出到/etl/out。现在,以下 pig 脚本失败并出现我在开头描述的错误。
a= LOAD '/etl/out' USING org.apache.pig.builtin.AvroStorage('hdfs:///etl/test.avsc')
b = FOREACH a GENERATE $0,$1,$2,$3,$4;
hdfs:///etl/test.avsc 中有 5 个字段(第 5 个是新字段)。
最佳答案
正如所怀疑的那样,问题实际上出在 AvroStorage 的使用方式上。用法是:a= LOAD '/etl/out' USING org.apache.pig.builtin.AvroStorage('test','-schemafile file:///etl/test.avsc') ,假设 test.avsc 在本地文件系统中。将 avsc 保留在 hdfs 中也应该可以工作,但我没有抽出时间来测试它。我不确定为什么没有关于这些的明确文档!!!!
关于java - Avro Schema Evolution With GenericData.Record - Mapreduce 过程,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/37081933/
我是一名优秀的程序员,十分优秀!