java - 使用自定义序列化器从 avro 读取时，RDD 中的运行时类型错误

Question

我正在尝试使用 Kryo 将 avro 文件中的数据读取到 RDD 中。我的代码编译得很好，但在运行时我得到了 ClassCastException 。这是我的代码的作用:

SparkConf conf = new SparkConf()...
conf.set("spark.serializer", KryoSerializer.class.getCanonicalName());
conf.set("spark.kryo.registrator", MyKryoRegistrator.class.getName());
JavaSparkContext sc = new JavaSparkContext(conf);

哪里MyKryoRegistrator为 MyCustomClass 注册一个序列化器:

public void registerClasses(Kryo kryo) {
    kryo.register(MyCustomClass.class, new MyCustomClassSerializer());
}

然后，我读取了我的数据文件:

JavaPairRDD<MyCustomClass, NullWritable> records =
                sc.newAPIHadoopFile("file:/path/to/datafile.avro",
                AvroKeyInputFormat.class, MyCustomClass.class, NullWritable.class,
                sc.hadoopConfiguration());
Tuple2<MyCustomClass, NullWritable> first = records.first();

这似乎工作正常，但使用调试器我可以看到，虽然 RDD 有一个 my.package.containing.MyCustomClass 的 kClassTag，但变量 first包含 Tuple2<AvroKey, NullWritable> ，不是Tuple2<MyCustomClass, NullWritable> !事实上，当执行以下行时:

System.out.println("Got a result, custom field is: " + first._1.getSomeCustomField());

我遇到异常:

java.lang.ClassCastException: org.apache.avro.mapred.AvroKey cannot be cast to my.package.containing.MyCustomClass

我做错了什么吗？即便如此，我不应该得到编译错误而不是运行时错误吗？

Answer 1

****************编辑****************

我设法从 avro 文件加载自定义对象并创建了 GitHub repository与代码。但是，如果 avro lib 无法将数据加载到自定义类中，它将返回 GenericData$Record 对象。在这种情况下，Spark Java API 不会检查对自定义类的分配，这就是为什么在尝试访问 AvroKey 的数据时只会收到 ClassCastException 的原因。这违反了数据安全保证。

****************编辑****************

对于尝试执行此操作的其他人，我有一个解决此问题的方法，但这不是正确的解决方案: 我创建了一个用于从 avro 文件读取 GenericData.Record 的类:

public class GenericRecordFileInputFormat extends FileInputFormat<GenericData.Record, NullWritable> {
    private static final Logger LOG = LoggerFactory.getLogger(GenericRecordFileInputFormat.class);

    /**
     * {@inheritDoc}
     */
    @Override
    public RecordReader<GenericData.Record, NullWritable> createRecordReader(
            InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        Schema readerSchema = AvroJob.getInputKeySchema(context.getConfiguration());
        if (null == readerSchema) {
            LOG.warn("Reader schema was not set. Use AvroJob.setInputKeySchema() if desired.");
            LOG.info("Using a reader schema equal to the writer schema.");
        }
        return new GenericDataRecordReader(readerSchema);
    }


    public static class GenericDataRecordReader extends RecordReader<GenericData.Record, NullWritable> {

        AvroKeyRecordReader<GenericData.Record> avroReader;

        public GenericDataRecordReader(Schema readerSchema) {
            super();
            avroReader = new AvroKeyRecordReader<>(readerSchema);
        }

        @Override
        public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
            avroReader.initialize(inputSplit, taskAttemptContext);
        }

        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException {
            return avroReader.nextKeyValue();
        }

        @Override
        public GenericData.Record getCurrentKey() throws IOException, InterruptedException {
            AvroKey<GenericData.Record> currentKey = avroReader.getCurrentKey();
            return currentKey.datum();
        }

        @Override
        public NullWritable getCurrentValue() throws IOException, InterruptedException {
            return avroReader.getCurrentValue();
        }

        @Override
        public float getProgress() throws IOException, InterruptedException {
            return avroReader.getProgress();
        }

        @Override
        public void close() throws IOException {
            avroReader.close();
        }
    }
}

然后我加载记录:

JavaRDD<GenericData.Record> records = sc.newAPIHadoopFile("file:/path/to/datafile.avro",
                GenericRecordFileInputFormat.class, GenericData.Record.class, NullWritable.class,
                sc.hadoopConfiguration()).keys();

然后，我使用接受 GenericData.Record 的构造函数将记录转换为自定义类。

再说一次 - 不漂亮，但有效。