apache/flink · error · IOException

Failed to deserialize CSV row '%s'.

Error message

Failed to deserialize CSV row '%s'.

What it means

Thrown by CsvRowDataDeserializationSchema.deserialize() when parsing the Kafka byte[] into a JsonNode or running the runtime converter fails and ignoreParseErrors is false. The raw CSV line is embedded in the message and the underlying Jackson/converter exception is chained. This is the classic 'dirty CSV row vs declared schema' failure for the csv format.

Source

Thrown at flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowDataDeserializationSchema.java:257

                    disabledFeatures.isEmpty()
                            ? EnumSet.noneOf(CsvParser.Feature.class)
                            : EnumSet.copyOf(disabledFeatures));
        }
    }

    @Override
    public RowData deserialize(@Nullable byte[] message) throws IOException {
        if (message == null) {
            return null;
        }
        try {
            final JsonNode root = objectReader.readValue(message);
            return (RowData) runtimeConverter.convert(root);
        } catch (Throwable t) {
            if (ignoreParseErrors) {
                return null;
            }
            throw new IOException(
                    String.format("Failed to deserialize CSV row '%s'.", new String(message)), t);
        }
    }

    @Override
    public boolean isEndOfStream(RowData nextElement) {
        return false;
    }

    @Override
    public TypeInformation<RowData> getProducedType() {
        return resultTypeInfo;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;

View on GitHub (pinned to 2f3c205e92)

Solutions

1. If occasional dirty rows are acceptable, set 'csv.ignore-parse-errors'='true' (rows are skipped/logged) — most common fix.
2. Otherwise correct the row or the producer so values match the declared column types.
3. Verify csv.field-delimiter / quote / escape options match the actual file format.
4. Reproduce offline: the offending CSV text is printed in the exception; feed it to a local CsvReader with the same schema.

Example fix

-- before
'format'='csv'

-- after
'format'='csv',
'csv.ignore-parse-errors'='true'

-- Opt-in tolerance when dirty rows are acceptable:
-- 'csv.ignore-parse-errors' = 'true'
-- For strict pipelines, sample and validate data first:
-- SELECT COUNT(*) FROM kafka_t /*+ OPTIONS('scan.startup.mode'='earliest') */ WHERE <type-check predicates>;

catch (IOException e) { if (ignoreParseErrors) { log.warn("skipping bad CSV row: {}", e.getMessage()); return null; } throw e; } — or simply enable the format's ignore-parse-errors option

When it happens

Trigger: A CSV line with wrong column count, non-numeric text in an INT column, malformed quoting, or a timestamp/date string that doesn't match the expected format; a schema change upstream (new column) while the Flink table schema stayed old; empty lines on the topic.

Common situations: Production topics fed by heterogeneous producers; late-arriving schema evolution; delimiter/quote misconfiguration (e.g. actual delimiter ';' but option default ','); locale-specific number formats.

Related errors

• Fail to deserialize at field: %s.
• Row too short: {}
• Line could not be parsed: '{}' ParserError {} Expect field
• Line could not be parsed: '{}' Expect field types: {} in fi
• value overflow

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/8a7ff539169a7bfa. Report an issue: GitHub.