{"record":{"id":"681e45565e13dab8","repo":"apache/beam","slug":"error-while-parsing-input-element","errorCode":null,"errorMessage":"Error while parsing input element","messagePattern":"Error while parsing input element","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"sdks/java/io/file-schema-transform/src/main/java/org/apache/beam/sdk/io/fileschematransform/FileWriteSchemaTransformFormatProviders.java","lineNumber":102,"sourceCode":"    private SerializableFunction<Row, OutputT> mapFn;\n    private Counter errorCounter;\n    private TupleTag<OutputT> outputTag;\n    private long errorsInBundle = 0L;\n\n    public BeamRowMapperWithDlq(\n        String name, SerializableFunction<Row, OutputT> mapFn, TupleTag<OutputT> outputTag) {\n      errorCounter = Metrics.counter(FileWriteSchemaTransformFormatProvider.class, name);\n      this.mapFn = mapFn;\n      this.outputTag = outputTag;\n    }\n\n    @ProcessElement\n    public void process(@DoFn.Element Row row, MultiOutputReceiver receiver) {\n      try {\n        receiver.get(outputTag).output(mapFn.apply(row));\n      } catch (Exception e) {\n        errorsInBundle += 1;\n        LOG.warn(\"Error while parsing input element\", e);\n        receiver\n            .get(ERROR_TAG)\n            .output(\n                Row.withSchema(ERROR_SCHEMA)\n                    .addValues(e.toString(), row.toString().getBytes(StandardCharsets.UTF_8))\n                    .build());\n      }\n    }\n\n    @FinishBundle\n    public void finish() {\n      errorCounter.inc(errorsInBundle);\n      errorsInBundle = 0L;\n    }\n  }\n\n  /**\n   * Applies common parameters from {@link FileWriteSchemaTransformConfiguration} to {@link","sourceCodeStart":84,"sourceCodeEnd":120,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/io/file-schema-transform/src/main/java/org/apache/beam/sdk/io/fileschematransform/FileWriteSchemaTransformFormatProviders.java#L84-L120","documentation":"FileWriteSchemaTransformFormatProviders' DoFn catches any exception thrown while applying the map function (e.g. a CsvI/O or Json format provider converting a Row to a record) during processElement. It increments the bundle error counter, logs this warning, and routes the failing element plus the exception message to the ERROR_TAG output instead of failing the pipeline. The error means the input Row could not be written/serialized in the requested format.","triggerScenarios":"mapFn.apply(row) throws for a given Row — e.g. malformed field values for the chosen CSV/JSON format provider, null in a non-nullable position, or charset/quote escaping failures.","commonSituations":"CSV provider given rows containing field values with unescaped delimiters; JSON provider receiving incompatible types; rows produced by FileReadSchemaTransform that don't match the writer's expected schema.","solutions":["Inspect the ERROR_TAG PCollection contents; it contains e.toString() and the offending serialized row.","Align the read schema with the write format (e.g. make nullable fields compatible with the provider).","Choose a different format provider or configure its options (delimiter, quoting) to fit the data.","Fix upstream data so fields conform, or filter/transform invalid rows before writing."],"exampleFix":"// before: rows with nulls written via CSV provider fail per-element\nFileIO.write().via(new SchemaTransformWrite(...)) // errors routed to ERROR_TAG\n// after: sanitize rows before writing\nrow -> Row.withSchema(schema).addValues(Optional.ofNullable(row.getValue(\"f\")).orElse(\"\")).build()","handlingStrategy":"fallback","validationCode":"// pre-check row compatibility with the target format provider\nboolean writable = row.getSchema().getFields().stream().allMatch(f -> row.getValue(f.getName()) != null || f.getType().getNullable());","typeGuard":null,"tryCatchPattern":"// consume the ERROR_TAG output instead of only the main output\nPCollection<Row> errors = result.get(ERROR_TAG);\nerrors.apply(\"LogBadRows\", ParDo.of(new LogFn<>())).setCoder(ErrorRowCoder.of());","preventionTips":["Match the read schema to the writer's format provider expectations (nullability, types).","Sanitize delimiters/quotes for CSV output before writing.","Always inspect the ERROR_TAG PCollection in tests."],"tags":["java","schema","serialization","beam"],"backgroundTag":"schema-validation-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}