{"record":{"id":"3883dcbd609aca88","repo":"apache/beam","slug":"attribute-fields-missing-attribute-names-not-found-in-schema","errorCode":null,"errorMessage":"Attribute fields {missing_attribute_names} not found in schema fields {schema_names}","messagePattern":"Attribute fields (.+?) not found in schema fields (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/yaml/yaml_io.py","lineNumber":513,"sourceCode":"    attributes = [attributes]\n  if attributes:\n    extra_fields.extend(attributes)\n  if attributes_map:\n    extra_fields.append(attributes_map)\n\n  def attributes_extractor(row):\n    if attributes_map:\n      attribute_values = dict(getattr(row, attributes_map))\n    else:\n      attribute_values = {}\n    if attributes:\n      attribute_values.update({attr: getattr(row, attr) for attr in attributes})\n    return attribute_values\n\n  schema_names = set(f.name for f in input_schema.fields)\n  missing_attribute_names = set(extra_fields) - schema_names\n  if missing_attribute_names:\n    raise ValueError(\n        f'Attribute fields {missing_attribute_names} '\n        f'not found in schema fields {schema_names}')\n\n  payload_schema = schema_pb2.Schema(\n      fields=[\n          field for field in input_schema.fields\n          if field.name not in extra_fields\n      ])\n  formatter = _create_formatter(format, schema, payload_schema)\n  return (\n      pcoll | beam.Map(\n          lambda row: beam.io.gcp.pubsub.PubsubMessage(\n              formatter(row), attributes_extractor(row)))\n      | beam.io.WriteToPubSub(\n          topic,\n          with_attributes=True,\n          id_label=id_attribute,\n          timestamp_attribute=timestamp_attribute))","sourceCodeStart":495,"sourceCodeEnd":531,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/yaml/yaml_io.py#L495-L531","documentation":"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.","triggerScenarios":"Calling write_to_pubsub with attributes (or attribute fields referenced via extra_fields) containing names absent from the input PCollection's schema.","commonSituations":"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.","solutions":["Align attribute names with actual input schema field names","Add the missing fields to the input schema (e.g. via a Map producing a schema'd row)","Inspect the input schema (pcoll.element_type) to confirm field names"],"exampleFix":"// before\nattributes: ['usrId']\n// after\nattributes: ['userId']  # matches input schema field","handlingStrategy":"validation","validationCode":"schema_names = {f.name for f in input_schema.fields}\nmissing = set(attributes) - schema_names\nif missing:\n    raise ValueError(f'Missing attribute fields: {missing}')","typeGuard":null,"tryCatchPattern":"try:\n    pcoll | yaml_io.write_to_pubsub(attributes=attrs, ...)\nexcept ValueError as e:\n    if 'not found in schema fields' in str(e):\n        pcoll = pcoll | beam.Map(add_missing_attribute_fields)","preventionTips":["Derive attribute list from the schema programmatically","Pin and test the upstream schema; watch for renames","Log the input element_type when configuring the sink"],"tags":["python","apache-beam","pubsub","schema","attributes"],"backgroundTag":"schema-validation-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}