JsonToRow { @Experimental(Kind.SCHEMAS) public static JsonToRowWithErrFn withExceptionReporting(Schema rowSchema) { return JsonToRowWithErrFn.forSchema(rowSchema); }  static PTransform<PCollection<String>, PCollection<Row>> withSchema(Schema rowSchema); static PTransform<PCollection<String>, PCollection<Row>> withSchemaAndNullBehavior(
      Schema rowSchema, NullBehavior nullBehavior); @Experimental(Kind.SCHEMAS) static JsonToRowWithErrFn withExceptionReporting(Schema rowSchema);  }
@Test @Category(NeedsRunner.class) public void testParsesRowsDeadLetterNoErrors() throws Exception { PCollection<String> jsonPersons = pipeline.apply("jsonPersons", Create.of(JSON_PERSON)); ParseResult results = jsonPersons.apply(JsonToRow.withExceptionReporting(PERSON_SCHEMA)); PCollection<Row> personRows = results.getResults(); PCollection<Row> errors = results.getFailedToParseLines(); PAssert.that(personRows).containsInAnyOrder(PERSON_ROWS); PAssert.that(errors).empty(); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testParsesRowsDeadLetterWithMissingFieldsNoErrors() throws Exception { PCollection<String> jsonPersons = pipeline.apply("jsonPersons", Create.of(JSON_PERSON_WITH_IMPLICIT_NULLS)); ParseResult results = jsonPersons.apply( JsonToRow.withExceptionReporting(PERSON_SCHEMA_WITH_NULLABLE_FIELD) .withNullBehavior(NullBehavior.ACCEPT_MISSING_OR_NULL)); PCollection<Row> personRows = results.getResults(); PCollection<Row> errors = results.getFailedToParseLines(); PAssert.that(personRows).containsInAnyOrder(PERSON_ROWS_WITH_NULLS); PAssert.that(errors).empty(); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testParsesErrorRowsDeadLetter() throws Exception { PCollection<String> jsonPersons = pipeline.apply("jsonPersons", Create.of(JSON_PERSON_WITH_ERR)); ParseResult results = jsonPersons.apply(JsonToRow.withExceptionReporting(PERSON_SCHEMA)); PCollection<Row> personRows = results.getResults(); PCollection<Row> errors = results.getFailedToParseLines(); PAssert.that(personRows).containsInAnyOrder(PERSON_ROWS); PAssert.that(errors) .containsInAnyOrder( row(JsonToRowWithErrFn.ERROR_ROW_SCHEMA, "{}"), row(JsonToRow.JsonToRowWithErrFn.ERROR_ROW_SCHEMA, "Is it 42?")); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testParsesErrorWithErrorMsgRowsDeadLetter() throws Exception { PCollection<String> jsonPersons = pipeline.apply("jsonPersons", Create.of(JSON_PERSON_WITH_ERR)); ParseResult results = jsonPersons.apply(JsonToRow.withExceptionReporting(PERSON_SCHEMA).withExtendedErrorInfo()); PCollection<Row> personRows = results.getResults(); PCollection<Row> errorsWithMsg = results.getFailedToParseLines(); PAssert.that(personRows).containsInAnyOrder(PERSON_ROWS); PAssert.that(errorsWithMsg) .containsInAnyOrder( row( JsonToRowWithErrFn.ERROR_ROW_WITH_ERR_MSG_SCHEMA, "{}", "Non-nullable field 'name' is not present in the JSON object."), row( JsonToRowWithErrFn.ERROR_ROW_WITH_ERR_MSG_SCHEMA, "Is it 42?", "Unable to parse Row")); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testParsesErrorWithErrorMsgWithRequireNullDeadLetter() throws Exception { PCollection<String> jsonPersons = pipeline.apply("jsonPersons", Create.of(JSON_PERSON_WITH_IMPLICIT_NULLS)); ParseResult results = jsonPersons.apply( JsonToRow.withExceptionReporting(PERSON_SCHEMA_WITH_NULLABLE_FIELD) .withExtendedErrorInfo() .withNullBehavior(NullBehavior.REQUIRE_NULL)); PCollection<Row> errorsWithMsg = results.getFailedToParseLines(); PAssert.that(errorsWithMsg) .containsInAnyOrder( row( JsonToRowWithErrFn.ERROR_ROW_WITH_ERR_MSG_SCHEMA, jsonPerson("person1", "80"), "Field 'knowsJavascript' is not present in the JSON object."), row( JsonToRowWithErrFn.ERROR_ROW_WITH_ERR_MSG_SCHEMA, jsonPerson("person3", "60"), "Field 'knowsJavascript' is not present in the JSON object.")); pipeline.run(); }
@Test @Category(NeedsRunner.class) public void testParsesErrorWithErrorMsgRowsDeadLetterWithCustomFieldNames() throws Exception { PCollection<String> jsonPersons = pipeline.apply("jsonPersons", Create.of(JSON_PERSON_WITH_ERR)); String customErrField = "CUSTOM_ERR"; String customLineField = "CUSTOM_LINE"; ParseResult results = jsonPersons.apply( JsonToRow.withExceptionReporting(PERSON_SCHEMA) .withExtendedErrorInfo() .setErrorField(customErrField) .setLineField(customLineField)); Schema customSchema = Schema.builder() .addField(customLineField, FieldType.STRING) .addField(customErrField, FieldType.STRING) .build(); PCollection<Row> personRows = results.getResults(); PCollection<Row> errorsWithMsg = results.getFailedToParseLines(); PAssert.that(personRows).containsInAnyOrder(PERSON_ROWS); PAssert.that(errorsWithMsg) .containsInAnyOrder( row(customSchema, "{}", "Non-nullable field 'name' is not present in the JSON object."), row(customSchema, "Is it 42?", "Unable to parse Row")); pipeline.run(); }