apache/beam · error · ValueError

Attribute fields {missing_attribute_names} not found in sche

Error message

Attribute fields {missing_attribute_names} not found in schema fields {schema_names}

What it means

write_to_pubsub maps attributes from fields of the input row. If any configured attribute (or extra field like the payload/attribute placeholders) is not present in the input schema, this ValueError lists the missing names and the available schema fields.

Source

Thrown at sdks/python/apache_beam/yaml/yaml_io.py:513

    attributes = [attributes]
  if attributes:
    extra_fields.extend(attributes)
  if attributes_map:
    extra_fields.append(attributes_map)

  def attributes_extractor(row):
    if attributes_map:
      attribute_values = dict(getattr(row, attributes_map))
    else:
      attribute_values = {}
    if attributes:
      attribute_values.update({attr: getattr(row, attr) for attr in attributes})
    return attribute_values

  schema_names = set(f.name for f in input_schema.fields)
  missing_attribute_names = set(extra_fields) - schema_names
  if missing_attribute_names:
    raise ValueError(
        f'Attribute fields {missing_attribute_names} '
        f'not found in schema fields {schema_names}')

  payload_schema = schema_pb2.Schema(
      fields=[
          field for field in input_schema.fields
          if field.name not in extra_fields
      ])
  formatter = _create_formatter(format, schema, payload_schema)
  return (
      pcoll | beam.Map(
          lambda row: beam.io.gcp.pubsub.PubsubMessage(
              formatter(row), attributes_extractor(row)))
      | beam.io.WriteToPubSub(
          topic,
          with_attributes=True,
          id_label=id_attribute,
          timestamp_attribute=timestamp_attribute))

View on GitHub (pinned to 12126d8942)

Solutions

1. Align attribute names with actual input schema field names
2. Add the missing fields to the input schema (e.g. via a Map producing a schema'd row)
3. Inspect the input schema (pcoll.element_type) to confirm field names

Example fix

// before
attributes: ['usrId']
// after
attributes: ['userId']  # matches input schema field

schema_names = {f.name for f in input_schema.fields}
missing = set(attributes) - schema_names
if missing:
    raise ValueError(f'Missing attribute fields: {missing}')

try:
    pcoll | yaml_io.write_to_pubsub(attributes=attrs, ...)
except ValueError as e:
    if 'not found in schema fields' in str(e):
        pcoll = pcoll | beam.Map(add_missing_attribute_fields)

When it happens

Trigger: Calling write_to_pubsub with attributes (or attribute fields referenced via extra_fields) containing names absent from the input PCollection's schema.

Common situations: Typos in attribute names ('user_id' vs 'userId'); schema changed upstream so a column was renamed/removed; mixing payload_field and attribute field names incorrectly.

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

• A pubsub message must not have more than 100 attributes.
• A pubsub message attribute key must not exceed 256 bytes.
• A pubsub message attribute value must not exceed 1024 bytes
• Expecting exactly one field, found {field_names}
• Schema must contain at least one field. Schema: %s

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