@Override
  public PTransform<PCollection<Row>, PCollectionTuple> buildTransform(
      FileWriteSchemaTransformConfiguration configuration, Schema schema) {

    return new PTransform<PCollection<Row>, PCollectionTuple>() {
      @Override
      public PCollectionTuple expand(PCollection<Row> input) {
        FileWriteSchemaTransformConfiguration.CsvConfiguration csvConfiguration =
            getCSVConfiguration(configuration);
        CSVFormat csvFormat =
            CSVFormat.Predefined.valueOf(csvConfiguration.getPredefinedCsvFormat()).getFormat();
        CsvIO.Write<Row> write =
            CsvIO.writeRows(configuration.getFilenamePrefix(), csvFormat).withSuffix(suffix);

        if (configuration.getCompression() != null) {
          write = write.withCompression(getCompression(configuration));
        }

        if (configuration.getNumShards() != null) {
          int numShards = getNumShards(configuration);
          // Python SDK external transforms do not support null values requiring additional check.
          if (numShards > 0) {
            write = write.withNumShards(numShards);
          }
        }

        if (!Strings.isNullOrEmpty(configuration.getShardNameTemplate())) {
          write = write.withShardTemplate(getShardNameTemplate(configuration));
        }

        if (!Strings.isNullOrEmpty(configuration.getFilenameSuffix())) {
          write = write.withSuffix(getFilenameSuffix(configuration));
        }

        WriteFilesResult<String> result = input.apply("Row to CSV", write);
        PCollection<String> output =
            result
                .getPerDestinationOutputFilenames()
                .apply("perDestinationOutputFilenames", Values.create());
        return PCollectionTuple.of(RESULT_TAG, output);
      }
    };
  }
@Test
  public void byteTypeNonRepeated() {
    System.out.println("BYTE TYPE NON REPEATED");
    String prefix = folder(ByteType.class, "byteTypeNonRepeated");
    String validField = "byte";
    PCollection<Row> input =
        writePipeline.apply(Create.of(DATA.byteTypeRows).withRowSchema(BYTE_TYPE_SCHEMA));
    PCollection<Row> modifiedInput = input.apply(Select.fieldNames(validField));
    Schema modifiedSchema = modifiedInput.getSchema();
    FileWriteSchemaTransformConfiguration configuration = buildConfiguration(prefix);
    PCollection<String> result =
        modifiedInput
            .apply(getProvider().buildTransform(configuration, modifiedSchema))
            .get(RESULT_TAG);
    PCollection<Long> numFiles = result.apply(Count.globally());
    PAssert.thatSingleton(numFiles).isEqualTo(1L);
    writePipeline.run().waitUntilFinish();

    PCollection<String> csv =
        readPipeline.apply(TextIO.read().from(configuration.getFilenamePrefix() + "*"));
    List<String> expected = new ArrayList<>();
    expected.add(validField);
    for (Row row : DATA.byteTypeRows) {
      expected.add(row.getByte(validField).toString());
    }
    PAssert.that(csv).containsInAnyOrder(expected);
    readPipeline.run();
  }