public static <T> ToJson<T> of() {
    return new ToJson<>();
  }
@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();
  }