public static CsvIOParse<Row> parseRows(Schema schema, CSVFormat csvFormat) {
    CsvIOParseHelpers.validateCsvFormat(csvFormat);
    CsvIOParseHelpers.validateCsvFormatWithSchema(csvFormat, schema);
    RowCoder coder = RowCoder.of(schema);
    CsvIOParseConfiguration.Builder<Row> builder = CsvIOParseConfiguration.builder();
    builder.setCsvFormat(csvFormat).setSchema(schema).setCoder(coder).setFromRowFn(row -> row);
    return CsvIOParse.<Row>builder().setConfigBuilder(builder).build();
  }
@Test
  public void parseRows() {
    Pipeline pipeline = Pipeline.create();
    PCollection<String> input =
        csvRecords(
            pipeline,
            "# This is a comment",
            "aBoolean,aDouble,aFloat,anInteger,aLong,aString",
            "true,1.0,2.0,3,4,foo",
            "N/A,6.0,7.0,8,9,bar",
            "false,12.0,14.0,8,24,\"foo\nbar\"",
            "true,1.0,2.0,3,4,foo$,bar");
    List<Row> want =
        Arrays.asList(
            Row.withSchema(NULLABLE_ALL_PRIMITIVE_DATA_TYPES_SCHEMA)
                .withFieldValue("aBoolean", true)
                .withFieldValue("aDouble", 1.0)
                .withFieldValue("aFloat", 2.0f)
                .withFieldValue("anInteger", 3)
                .withFieldValue("aLong", 4L)
                .withFieldValue("aString", "foo")
                .build(),
            Row.withSchema(NULLABLE_ALL_PRIMITIVE_DATA_TYPES_SCHEMA)
                .withFieldValue("aBoolean", null)
                .withFieldValue("aDouble", 6.0)
                .withFieldValue("aFloat", 7.0f)
                .withFieldValue("anInteger", 8)
                .withFieldValue("aLong", 9L)
                .withFieldValue("aString", "bar")
                .build(),
            Row.withSchema(NULLABLE_ALL_PRIMITIVE_DATA_TYPES_SCHEMA)
                .withFieldValue("aBoolean", false)
                .withFieldValue("aDouble", 12.0)
                .withFieldValue("aFloat", 14.0f)
                .withFieldValue("anInteger", 8)
                .withFieldValue("aLong", 24L)
                .withFieldValue("aString", "foo\nbar")
                .build(),
            Row.withSchema(NULLABLE_ALL_PRIMITIVE_DATA_TYPES_SCHEMA)
                .withFieldValue("aBoolean", true)
                .withFieldValue("aDouble", 1.0)
                .withFieldValue("aFloat", 2.0f)
                .withFieldValue("anInteger", 3)
                .withFieldValue("aLong", 4L)
                .withFieldValue("aString", "foo,bar")
                .build());

    CsvIOParse<Row> underTest =
        CsvIO.parseRows(NULLABLE_ALL_PRIMITIVE_DATA_TYPES_SCHEMA, csvFormat());
    CsvIOParseResult<Row> result = input.apply(underTest);
    PAssert.that(result.getOutput()).containsInAnyOrder(want);
    PAssert.that(result.getErrors()).empty();

    pipeline.run();
  }