CsvToAvro { public static String getSchema(String schemaPath) throws IOException { ReadableByteChannel chan = FileSystems.open(FileSystems.matchNewResource( schemaPath, false)); try (InputStream stream = Channels.newInputStream(chan)) { BufferedReader streamReader = new BufferedReader(new InputStreamReader(stream, "UTF-8")); StringBuilder dataBuilder = new StringBuilder(); String line; while ((line = streamReader.readLine()) != null) { dataBuilder.append(line); } return dataBuilder.toString(); } }  static String getSchema(String schemaPath); static void checkFieldTypes(Schema schema); static void runCsvToAvro(SampleOptions options); static void main(String[] args);  }
@Test public void testCsvToAvro() throws Exception { SampleOptions options = TestPipeline.testingPipelineOptions().as(SampleOptions.class); options.setAvroSchema("gs: options.setInputFile("gs: String schemaJson = getSchema(options.getAvroSchema()); Schema schema = new Schema.Parser().parse(schemaJson); final List<GenericRecord> expectedResult = new ArrayList<>(); GenericRecord genericRecordOne = new GenericData.Record(schema); genericRecordOne.put("first_name", "frank"); genericRecordOne.put("last_name", "natividad"); genericRecordOne.put("age", 1); expectedResult.add(genericRecordOne); GenericRecord genericRecordTwo = new GenericData.Record(schema); genericRecordTwo.put("first_name", "Karthi"); genericRecordTwo.put("last_name", "thyagarajan"); genericRecordTwo.put("age", 3); expectedResult.add(genericRecordTwo); final PCollection<GenericRecord> avroDataCollection = pipeline.apply("Read CSV files", TextIO.read().from(options.getInputFile())) .apply("Convert CSV to Avro formatted data", ParDo.of( new CsvToAvro.ConvertCsvToAvro(schemaJson, options.getCsvDelimiter()))) .setCoder(AvroCoder.of(GenericRecord.class, schema)); PAssert.that(avroDataCollection).containsInAnyOrder(expectedResult); pipeline.run().waitUntilFinish(); }