apache/beam · error · ValueError

Can only use expressions on a schema'd input.

Error message

Can only use expressions on a schema'd input.

What it means

In `_as_callable_for_pcoll`, if the input PCollection's element type cannot be turned into a named schema (no schema), expressions—which need to reference named fields—cannot work. When the fn_spec looks like an expression, the transform raises this ValueError instead of silently proceeding.

Solutions

1. Ensure the input has a schema: insert a parse/convert step (e.g. json.Parse or a MapToFields with explicit output_type defining the fields) before the expression-based transform.
2. If the input is genuinely untyped, use a plain callable/lambda (`callable:` or `path:` config) instead of an `expression:`.
3. Verify the upstream transform preserves the Beam schema (check with a Print or schema inspection).

Example fix

- type: MapToFields
  input: RawText   # strings, no schema -> fails
# after: parse first
- type: MapToFields
  input: ParsedJson
  config:
    input_type:
      id: string
    fields:
      id: "id.upper()"

# ensure input has a schema before using expressions
from apache_beam.typehints.schemas import named_fields_from_element_type
fields = named_fields_from_element_type(pcoll.element_type)  # raises if untyped

def is_schema_apcoll(pcoll) -> bool:
    try:
        named_fields_from_element_type(pcoll.element_type)
        return True
    except (TypeError, ValueError):
        return False

try:
    out = apply_map_to_fields(pcoll, cfg)
except ValueError as e:
    if "expressions on a schema'd input" in str(e):
        pcoll = parse_to_schema(pcoll)
        out = apply_map_to_fields(pcoll, cfg)
    else:
        raise

When it happens

Trigger: Using MapToFields (or Filter/Partition/AssignTimestamps via _as_callable_for_pcoll) with an `expression:` config on a PCollection whose elements are untyped (e.g. plain strings, ints, or dicts rather than schema'd rows), such as output of a raw Read or non-schema transform.

Common situations: Applying a YAML MapToFields transform directly after a text/byte source like ReadFromText, or after a transform that loses the schema, while using field-referencing expressions.

Understand the failure class

Background: "is not a compatible type" / "cannot merge" errors: when a value's type doesn't match what the library requires — this error's family across 65 libraries.

Related errors

• violates schema
• The input to this transform does not appear to be an error…
• Unknown grouping columns
• Unknown or unsupported atomic type
• Unrecognized type_info

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

Appendix: source

Thrown at sdks/python/apache_beam/yaml/yaml_mapping.py:381

        validator(getattr(row, name))
        for (name, validator) in validators.items())
  else:
    raise ValueError(f"Unrecognized type_info: {type_info!r}")


def _as_callable_for_pcoll(
    pcoll,
    fn_spec: Union[str, dict[str, str]],
    msg: str,
    language: Optional[str]):
  if language == 'javascript':
    options.YamlOptions.check_enabled(pcoll.pipeline, 'javascript')

  try:
    input_schema = dict(named_fields_from_element_type(pcoll.element_type))
  except (TypeError, ValueError) as exn:
    if is_expr(fn_spec):
      raise ValueError("Can only use expressions on a schema'd input.") from exn
    input_schema = {}  # unused

  if isinstance(fn_spec, str) and fn_spec in input_schema:
    return lambda row: getattr(row, fn_spec)
  else:
    return _as_callable(
        list(input_schema.keys()), fn_spec, msg, language, input_schema)


def _as_callable(original_fields, expr, transform_name, language, input_schema):
  if isinstance(expr, str):
    expr = {'expression': expr}

  # Extract original type from upstream pcoll when doing simple mappings
  original_type = input_schema.get(expr.get('expression'), None)
  if expr in original_fields:
    language = "python"

View on GitHub (pinned to 12126d8942)