apache/beam · error · TypeError

Element type must be compatible with Beam Schemas (https://b

Error message

Element type must be compatible with Beam Schemas (https://beam.apache.org/documentation/programming-guide/#schemas) for batch type pd.DataFrame

What it means

DataFrameBatchConverter.from_typehints requires the element type to be a Beam schema-compatible row type because batches are pd.DataFrame, whose columns must correspond to schema fields. If the element_type is not already a RowTypeConstraint and RowTypeConstraint.from_user_type fails to derive one (e.g. the user type is not a NamedTuple or a class annotated with @beam_typehints row schema), this TypeError is raised.

Source

Thrown at sdks/python/apache_beam/typehints/pandas_type_compatibility.py:168

class DataFrameBatchConverter(BatchConverter):
  def __init__(
      self,
      element_type: RowTypeConstraint,
  ):
    super().__init__(pd.DataFrame, element_type)
    self._columns = [name for name, _ in element_type._fields]

  @staticmethod
  def from_typehints(element_type,
                     batch_type) -> Optional['DataFrameBatchConverter']:
    assert batch_type == pd.DataFrame

    if not isinstance(element_type, RowTypeConstraint):
      element_type = RowTypeConstraint.from_user_type(element_type)
      if element_type is None:
        raise TypeError(
            "Element type must be compatible with Beam Schemas ("
            "https://beam.apache.org/documentation/programming-guide/#schemas) "
            "for batch type pd.DataFrame")

    index_columns = [
        field_name
        for (field_name, field_options) in element_type._field_options.items()
        if any(key == INDEX_OPTION_NAME for key, value in field_options)
    ]

    if index_columns:
      return DataFrameBatchConverterKeepIndex(element_type, index_columns)
    else:
      return DataFrameBatchConverterDropIndex(element_type)

  def _get_series(self, batch: pd.DataFrame):
    raise NotImplementedError

View on GitHub (pinned to 12126d8942)

Solutions

1. Change the element type to a schema-compatible row type, e.g. a typing.NamedTuple or a class whose fields are Beam schema-compatible.
2. Alternatively use batch_type=pd.Series, which supports scalar (non-row) element types.
3. If the user type is a plain class, add type annotations to its fields so RowTypeConstraint.from_user_type can build a schema.
4. Verify element_type is not None/dynamic before constructing the converter and fail with a clearer message.

Example fix

// before
class MyElement:  # no schema
    pass
converter = BatchConverter.from_typehints(element_type=MyElement, batch_type=pd.DataFrame)
// after
import typing
import pandas as pd
class MyElement(typing.NamedTuple):
    x: int
    y: str
converter = BatchConverter.from_typehints(element_type=MyElement, batch_type=pd.DataFrame)

from apache_beam.typehints.row_type import RowTypeConstraint
def is_df_batch_compatible(element_type, batch_type) -> bool:
    import pandas as pd
    if batch_type == pd.Series:
        return True
    return isinstance(element_type, RowTypeConstraint) or RowTypeConstraint.from_user_type(element_type) is not None

def is_row_schema_type(element_type) -> bool:
    from apache_beam.typehints.row_type import RowTypeConstraint
    return isinstance(element_type, RowTypeConstraint) or RowTypeConstraint.from_user_type(element_type) is not None

if not is_row_schema_type(element_type):
    # switch to Series batching or fix the element type before constructing
    batch_type = pd.Series
converter = create_pandas_batch_converter(element_type=element_type, batch_type=batch_type)

When it happens

Trigger: Calling DataFrameBatchConverter.from_typehints (via create_pandas_batch_converter or BatchConverter.from_typehints) with batch_type=pd.DataFrame and an element_type that cannot be converted to a Beam schema: a primitive like int, a plain dict-typed hint without schema, or an unannotated class.

Common situations: Batching scalar/primitive elements into DataFrames; forgetting to decorate element classes with typing.NamedTuple or beam schema annotations; using Optional[Any] element hints in pipelines that also use beam.BatchElements.

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

• Only dataframes with single rows are supported.
• batch type must be pd.Series or pd.DataFrame
• concat(ignore_index)
• concat(levels)
• min_batch_size must be >= 1, got {min_batch_size}

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