wholefile-transformer-lib/src/main/java/com/streamsets/pipeline/stage/processor/transformer/WholeFileTransformerProcessor.java
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());
  }
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());
  }
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
@param sourceFileName the source Avro file name
@param fileReader the {@link org.apache.avro.file.DataFileStream} Avro file reader
@param tempParquetFile the {@link java.nio.file.Path} temporary parquet file path
Convert Avro record to Parquet