public static PTransform<PCollection<String>, PCollection<Row>> withSchemaAndNullBehavior(
      Schema rowSchema, NullBehavior nullBehavior) {
    RowJson.verifySchemaSupported(rowSchema);
    return new JsonToRowFn(rowSchema, nullBehavior);
  }
@Test
  @Category(NeedsRunner.class)
  public void testParsesRowsMissing() throws Exception {

    PCollection<String> jsonPersons =
        pipeline.apply("jsonPersons", Create.of(JSON_PERSON_WITH_IMPLICIT_NULLS));

    PCollection<Row> personRows =
        jsonPersons
            .apply(
                ((JsonToRowFn)
                    JsonToRow.withSchemaAndNullBehavior(
                        PERSON_SCHEMA_WITH_NULLABLE_FIELD, NullBehavior.ACCEPT_MISSING_OR_NULL)))
            .setRowSchema(PERSON_SCHEMA_WITH_NULLABLE_FIELD);

    PAssert.that(personRows).containsInAnyOrder(PERSON_ROWS_WITH_NULLS);

    pipeline.run();
  }