[jira] [Created] (FLINK-21406) Add AvroParquetFileRecordFormat

[Shang Yuanchun (Jira)]

Chesnay Schepler created FLINK-21406:
----------------------------------------

             Summary: Add AvroParquetFileRecordFormat
                 Key: FLINK-21406
                 URL: https://issues.apache.org/jira/browse/FLINK-21406
             Project: Flink
          Issue Type: New Feature
          Components: API / DataStream, Formats (JSON, Avro, Parquet, ORC, SequenceFile)
            Reporter: Chesnay Schepler
             Fix For: 1.13.0


There is currently no easy way to read avro GenericRecords from parquet via the new {{FileSource}}.
While helping out a user I started writing FileRecordFormat that could do that, but it requires some finalization.

The implementation is similar to our ParquetAvroWriters class, in that we just wrap some parquet classes and bridge our FileSystem with their IO abstraction.

The main goal was to have a format that reads data through our FileSystems, and not work directly against Hadoop to prevent a ClassLoader leak from the S3AFileSystem (threads in a thread pool can keep references to the user classloader).

According to the user it appears to be working, but it will need some cleanup, ideally support for specific records, support for checkpointing (which should be fairly easy I believe), maybe splitting files (not sure whether this works properly with Parquet) and of course tests + documentation.

{code}
public class ParquetAvroFileRecordFormat implements FileRecordFormat<GenericRecord> {
    private final transient Schema schema;

    public ParquetAvroFileRecordFormat(Schema schema) {
        this.schema = schema;
    }

    @Override
    public Reader<GenericRecord> createReader(
            Configuration config, Path filePath, long splitOffset, long splitLength)
            throws IOException {

        final FileSystem fs = filePath.getFileSystem();
        final FileStatus status = fs.getFileStatus(filePath);
        final FSDataInputStream in = fs.open(filePath);

        return new MyReader(
                AvroParquetReader.<GenericRecord>builder(new InputFileWrapper(in, status.getLen()))
                        .withDataModel(GenericData.get())
                        .build());
    }

    @Override
    public Reader<GenericRecord> restoreReader(
            Configuration config,
            Path filePath,
            long restoredOffset,
            long splitOffset,
            long splitLength) {
        // not called if checkpointing isn't used
        return null;
    }

    @Override
    public boolean isSplittable() {
        // let's not worry about this for now
        return false;
    }

    @Override
    public TypeInformation<GenericRecord> getProducedType() {
        return new GenericRecordAvroTypeInfo(schema);
    }

    private static class MyReader implements FileRecordFormat.Reader<GenericRecord> {

        private final ParquetReader<GenericRecord> parquetReader;

        private MyReader(ParquetReader<GenericRecord> parquetReader) {
            this.parquetReader = parquetReader;
        }

        @Nullable
        @Override
        public GenericRecord read() throws IOException {
            return parquetReader.read();
        }

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

    private static class InputFileWrapper implements InputFile {

        private final FSDataInputStream inputStream;
        private final long length;

        private InputFileWrapper(FSDataInputStream inputStream, long length) {
            this.inputStream = inputStream;
            this.length = length;
        }

        @Override
        public long getLength() {
            return length;
        }

        @Override
        public SeekableInputStream newStream() {
            return new SeekableInputStreamAdapter(inputStream);
        }
    }

    private static class SeekableInputStreamAdapter extends DelegatingSeekableInputStream {

        private final FSDataInputStream inputStream;

        private SeekableInputStreamAdapter(FSDataInputStream inputStream) {
            super(inputStream);
            this.inputStream = inputStream;
        }

        @Override
        public long getPos() throws IOException {
            return inputStream.getPos();
        }

        @Override
        public void seek(long newPos) throws IOException {
            inputStream.seek(newPos);
        }
    }
}
{code}



--
This message was sent by Atlassian Jira
(v8.3.4#803005)