ToJson extends PTransform<PCollection<T>, PCollection<String>> { public static <T> ToJson<T> of() { return new ToJson<>(); } private  ToJson(); static ToJson<T> of(); @Override PCollection<String> expand(PCollection<T> rows);  }
@Test @Category(NeedsRunner.class) public void testSerializesParseableJson() throws Exception { PCollection<Person> persons = pipeline.apply( "jsonPersons", Create.of( Person.of("person1", 80, true), Person.of("person2", 70, false), Person.of("person3", 60, true), Person.of("person4", 50, false), Person.of("person5", 40, true))); Schema personSchema = Schema.builder() .addStringField("name") .addInt32Field("height") .addBooleanField("knowsJavascript") .build(); PCollection<Row> personRows = persons.apply(ToJson.of()).apply(JsonToRow.withSchema(personSchema)); PAssert.that(personRows) .containsInAnyOrder( row(personSchema, "person1", 80, true), row(personSchema, "person2", 70, false), row(personSchema, "person3", 60, true), row(personSchema, "person4", 50, false), row(personSchema, "person5", 40, true)); pipeline.run(); }