Hi All, I am trying to read Avro data from Kafka which uses schema registry and write it to s3 in Parquet format. I am facing this issue currently

Caused by: org.apache.parquet.io.ParquetEncodingException: empty fields are illegal, the field should be ommited completely instead
	at org.apache.parquet.io.MessageColumnIO$MessageColumnIORecordConsumer.endField(MessageColumnIO.java:329)
	at org.apache.flink.formats.parquet.row.ParquetRowDataWriter$ArrayWriter.write(ParquetRowDataWriter.java:420)
	at org.apache.flink.formats.parquet.row.ParquetRowDataWriter$RowWriter.write(ParquetRowDataWriter.java:456)

As far as I have checked we can avoid this by using AvroParquetWriter and providing the config

parquet.avro.write-old-list-structure

as false. Can anybody here help on how this can be achieved ?

Hi @Alokh P, Were you able to solve this?

I would also like to know how this was done

Initially, I was creating TumblingWindows in the SQL query and later realised that I don’t need that. Then I removed the same as I have partitionedBy clause in the sink table. From then on I’m facing this error. Not sure what has went wrong here.

@Raghunadh Nittala I solved it using the datastream API, was not able to figure out how to do it with table API. Probably the same works in Table API as well as I am modifying the Parquet class itself. Below is the method that generates the sink for me.

private static FileSink<GenericRecord> getFileSink(Schema schema) {
    return FileSink
            .forBulkFormat(
                    new Path(ns.getString(SINK_S3_PATH)),
                    AvroParquetWriters.forGenericRecord(schema))
            .disableCompact()
            .build();
}

In the current API, I was not able to figure out how to pass the required configurations to the ParquetWriter. I got this running in a hacky way by modifying the build method to hardcode my required config as below in org.apache.parquet.hadoop.ParquetWriter class.

public ParquetWriter<T> build() throws IOException {
    conf.set(WRITE_OLD_LIST_STRUCTURE, "false");
    codecName = CompressionCodecName.SNAPPY;
    if (file != null) {
        return new ParquetWriter<>(file,
                mode, getWriteSupport(conf), codecName, rowGroupSize, enableValidation, conf,
                maxPaddingSize, encodingPropsBuilder.build(), encryptionProperties);
    } else {
        return new ParquetWriter<>(HadoopOutputFile.fromPath(path, conf),
                mode, getWriteSupport(conf), codecName,
                rowGroupSize, enableValidation, conf, maxPaddingSize,
                encodingPropsBuilder.build(), encryptionProperties);
    }
}

The proper way would be to modify the API to accept configurations and pass it on to the parquet writer accordingly. As mentioned earlier I could not find an existing way to do so hence chose to do this. Do let me know incase anybody is aware of a cleaner way to do this.

Thanks @Alokh P for your reply, sorry for replying late.