private void writeParquet ( String sourceFileName , DataFileStream < GenericRecord > fileReader , Path tempParquetFile ) throws StageException { long recordCount = 0 ; GenericRecord avroRecord ; Schema schema = fileReader . getSchema ( ) ; LOG . debug ( "Start reading input file : {}" , sourceFileName ) ; try { // initialize parquet writer Configuration jobConfiguration = new Configuration ( ) ; String compressionCodecName = compressionElEval . eval ( variables , jobConfig . avroParquetConfig . compressionCodec , String . class ) ; jobConfiguration . set ( AvroParquetConstants . COMPRESSION_CODEC_NAME , compressionCodecName ) ; jobConfiguration . setInt ( AvroParquetConstants . ROW_GROUP_SIZE , jobConfig . avroParquetConfig . rowGroupSize ) ; jobConfiguration . setInt ( AvroParquetConstants . PAGE_SIZE , jobConfig . avroParquetConfig . pageSize ) ; jobConfiguration . setInt ( AvroParquetConstants . DICTIONARY_PAGE_SIZE , jobConfig . avroParquetConfig . dictionaryPageSize ) ; jobConfiguration . setInt ( AvroParquetConstants . MAX_PADDING_SIZE , jobConfig . avroParquetConfig . maxPaddingSize ) ; // Parquet writer ParquetWriter . Builder builder = AvroToParquetConverterUtil . initializeWriter ( new org . apache . hadoop . fs . Path ( tempParquetFile . toString ( ) ) , schema , jobConfiguration ) ; parquetWriter = builder . build ( ) ; while ( fileReader . hasNext ( ) ) { avroRecord = fileReader . next ( ) ; parquetWriter . write ( avroRecord ) ; recordCount ++ ; } parquetWriter . close ( ) ; } catch ( IOException ex ) { throw new TransformerStageCheckedException ( Errors . CONVERT_08 , sourceFileName , recordCount , ex ) ; } LOG . debug ( "Finished writing {} records to {}" , recordCount , tempParquetFile . getFileName ( ) ) ; }
Convert Avro record to Parquet