Parquet Writer

ParquetWriter.kt

class ParquetWriter<T : SpecificRecord>(private val outputFile: String, schema: Schema, conf: Configuration) {
    private var closed = false

    private val writer by lazy {
        val timeSupport = GenericData().also {
            it.addLogicalTypeConversion(TimeConversions.DateConversion())
        }

        AvroParquetWriter.builder<T>(HadoopOutputFile.fromPath(Path(outputFile), conf))
            .withSchema(schema)
            .withDataModel(timeSupport)
            .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
            .withCompressionCodec(CompressionCodecName.SNAPPY)
            .build()
    }

    fun write(records: Collection<T>) {
        if (closed) throw IllegalStateException("The writer is closed and cannot be used to write anymore")

        records.forEach {
            writer.write(it)
        }
    }

    fun close() {
        writer.close()
        closed = true
    }
}