AvroToCsv { private static String getSchema(String schemaPath) throws IOException { ReadableByteChannel channel = FileSystems.open(FileSystems.matchNewResource( schemaPath, false)); try (InputStream stream = Channels.newInputStream(channel)) { 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 void checkFieldTypes(Schema schema); static void runAvroToCsv(SampleOptions options); static void main(String[] args);  }
@Test public void testAvroToCsv() throws Exception { final List<String> expectedList = Arrays.asList("frank,natividad,1", "Karthi,thyagarajan,3"); SampleOptions options = TestPipeline.testingPipelineOptions().as(SampleOptions.class); options.setAvroSchema("gs: options.setInputFile("gs: String schemaJson = getSchema(options.getAvroSchema()); final PCollection<String> csvDataCollection = pipeline .apply("Read input", AvroIO.readGenericRecords(schemaJson).from(options.getInputFile())) .apply("Convert Avro to CSV formatted data", ParDo.of(new AvroToCsv.ConvertAvroToCsv(schemaJson, options.getCsvDelimiter()))); System.out.println(csvDataCollection); PAssert.that(csvDataCollection).containsInAnyOrder(expectedList); pipeline.run().waitUntilFinish(); }