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());
  }
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