apache/beam · error · ValueError

Attribute fields not found in schema fields

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.

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 attribute key must not exceed 256 bytes.
• A pubsub message attribute value must not exceed 1024 bytes
• A pubsub message must not have more than 100 attributes.
• Expecting exactly one field, found
• A pubsub message data field must not exceed 10MB

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

Appendix: 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)