public static PTransform<PCollection<String>, PCollection<Row>> withSchema(Schema rowSchema) {
    RowJson.verifySchemaSupported(rowSchema);
    return new JsonToRowFn(rowSchema, NullBehavior.ACCEPT_MISSING_OR_NULL);
  }
@Test
  @Category(NeedsRunner.class)
  public void testParsesRows() throws Exception {
    Schema personSchema =
        Schema.builder()
            .addStringField("name")
            .addInt32Field("height")
            .addBooleanField("knowsJavascript")
            .build();

    PCollection<String> jsonPersons =
        pipeline.apply(
            "jsonPersons",
            Create.of(
                jsonPerson("person1", "80", "true"),
                jsonPerson("person2", "70", "false"),
                jsonPerson("person3", "60", "true"),
                jsonPerson("person4", "50", "false"),
                jsonPerson("person5", "40", "true")));

    PCollection<Row> personRows =
        jsonPersons.apply(JsonToRow.withSchema(personSchema)).setRowSchema(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();
  }