[HIVE-28628] override parameter to Avro DataFileWriter to avoid to many flush, 1 per block of 64k

• Type: Improvement
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Component/s: Avro
• Security Level: Public (Viewable by anyone)
• Assignee: Unassigned
• Reporter: Arnaud Nauwynck
• Created: 6 days ago 09:13
• Updated: 6 days ago 09:13

Description

when saving big Avro file with spark, it may be extremely slow because of the "flush()" that are called many times, for each avro block ~64k of bytes.

Here is the corresponding jira ticket in spark:
https://issues.apache.org/jira/browse/SPARK-50326

This is especially slow on a an Hadoop VirtualFileSystem for which the flush() is really doing many things internally, in addition to reconnecting a Https connection (example: Azure Storage).

Currently, the code that implements the Avro format support is from hive-exec jar.
At the lowest level, it is doing

https://github.com/apache/hive/blob/master/ql/src/java/org/apache/hadoop/hive/ql/io/avro/AvroContainerOutputFormat.java#L71

public class AvroContainerOutputFormat {

    public FileSinkOperator.RecordWriter getHiveRecordWriter(JobConf jobConf, Path path, Class<? extends Writable> valueClass, boolean isCompressed, Properties properties, Progressable progressable) throws IOException {
        Schema schema;
        try {
            schema = AvroSerdeUtils.determineSchemaOrThrowException(jobConf, properties);
        } catch (AvroSerdeException var13) {
            AvroSerdeException e = var13;
            throw new IOException(e);
        }

        GenericDatumWriter<GenericRecord> gdw = new GenericDatumWriter(schema);
        DataFileWriter<GenericRecord> dfw = new DataFileWriter(gdw);
        if (isCompressed) {
            int level = jobConf.getInt("avro.mapred.deflate.level", -1);
            String codecName = jobConf.get("avro.output.codec", "deflate");
            CodecFactory factory = codecName.equals("deflate") ? CodecFactory.deflateCodec(level) : CodecFactory.fromString(codecName);
            dfw.setCodec(factory);
        }

        dfw.create(schema, path.getFileSystem(jobConf).create(path));
        return new AvroGenericRecordWriter(dfw);
    }

As you can see only 2 options are used from the spark->hadoop jobConf -> hive-exec to the Avro library : the schema, and the codec compression.

The Avro class "DataFileWriter" supports 2 other importants attributes, but the default value are suited only for doing small files (or kafka streaming):

public class DataFileWriter<D> implements Closeable, Flushable {
    private int syncInterval = 64000;
    private boolean flushOnEveryBlock = true;

The proposed changed is to override (upgrade?) the class in hive-exec so that these 2 attributes could be configured with better values.

DataFileWriter<GenericRecord> dfw = new DataFileWriter(gdw);

Boolean flushOnEveryBlock = .... Default true ... may change to false from optional conf
               dfw.setFlushOnEveryBlock(false);

int syncInterval =  8388608; // 8mo  ...  default 64000, and max allowed 1073741824=1go  ... may change from optional conf
               dfw.setFlushOnEveryBlock(syncInterval);