apache/beam · critical · IllegalArgumentException

requires an input Schema. Note that only Row or user classe

Error message

 requires an input Schema. Note that only Row or user classes are supported. Consider using TextIO or FileIO directly when writing primitive types

What it means

CsvIO.Write's expand() requires the input PCollection to have a Beam Schema, because CSV columns are derived from schema fields. Primitive types (String, Integer, byte[]) have no schema, so writing them as CSV is ambiguous and the transform throws IllegalArgumentException, pointing users to TextIO/FileIO.

Source

Thrown at sdks/java/io/csv/src/main/java/org/apache/beam/sdk/io/csv/CsvIO.java:662

      abstract Write<T> autoBuild();

      final Write<T> build() {

        if (getCSVFormat().getHeaderComments() != null) {
          checkArgument(
              getCSVFormat().isCommentMarkerSet(),
              "CSVFormat withCommentMarker required when withHeaderComments");
        }

        return autoBuild();
      }
    }

    @Override
    public WriteFilesResult<String> expand(PCollection<T> input) {
      if (!input.hasSchema()) {
        throw new IllegalArgumentException(
            String.format(
                "%s requires an input Schema. Note that only Row or user classes are supported. Consider using TextIO or FileIO directly when writing primitive types",
                Write.class.getName()));
      }

      Schema schema = input.getSchema();

      RowCoder rowCoder = RowCoder.of(schema);

      PCollection<Row> rows =
          input
              .apply("To Rows", MapElements.into(rows()).via(input.getToRowFunction()))
              .setCoder(rowCoder);

      CSVFormat csvFormat = buildHeaderFromSchemaIfNeeded(getCSVFormat(), schema);

      TextIO.Write write = getTextIOWrite();

View on GitHub (pinned to 12126d8942)

Solutions

Provide a schema: use CsvIO.<T>write(...).withFormatFunction(MyClass::toRow) or ensure the element class is a supported @DefaultSchema/annotated class.
For primitive types, write with TextIO.write() or FileIO instead of CsvIO.
Call input.setSchema(...) / apply a schema-transform upstream so the PCollection carries a Schema before CsvIO.write.

Example fix

// before
pipeline.apply(Create.of("a,b,c")).apply(CsvIO.write(path));
// after
pipeline.apply(Create.of(rowOrUserClassElements)).apply(CsvIO.write(path));
// or for raw text: .apply(TextIO.write().to(path));

Defensive patterns

Strategy: type-guard

Validate before calling

if (!input.hasSchema()) {
  throw new IllegalStateException("CsvIO.write requires a schema; use withFormatFunction or TextIO for primitives");
}
input.apply(CsvIO.write(path));

Type guard

boolean csvWritable(PCollection<?> input) { return input != null && input.hasSchema(); }

Try / catch

try { input.apply(CsvIO.write(path)); } catch (IllegalArgumentException e) { input.apply(TextIO.write().to(path)); }

Prevention

Only feed Row or schema-annotated user classes into CsvIO.write.
For raw text or primitive PCsollections, use TextIO/FileIO directly.
Check hasSchema() during pipeline assembly (graph-building time) to fail early.

When it happens

Trigger: Applying CsvIO.write(...) to a PCollection<String>, PCollection<Integer>, or any element type created without withFormatFunction/setSchema, so input.hasSchema() is false.

Common situations: Piping raw Strings (e.g. pre-formatted CSV lines) into CsvIO; upgrading a TextIO pipeline to CsvIO; creating a PCollection via Create.of("a,b") without a schema; using a user class without registering a schema via withFormatFunction.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/c25c3ad857f881dc. Report an issue: GitHub.