{"record":{"id":"77d2de177bef62ce","repo":"apache/beam","slug":"cannot-encode-payload-for-writetopubsub-expected-valid","errorCode":null,"errorMessage":"Cannot encode payload for WriteToPubSub. Expected valid string or bytes object, got {repr(output)} of type {type(output)}.","messagePattern":"Cannot encode payload for WriteToPubSub\\. Expected valid string or bytes object, got (.+?) of type (.+?)\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/yaml/yaml_io.py","lineNumber":296,"sourceCode":"\n  if format.islower():\n    format = format.upper()\n\n  if format == 'RAW':\n    if schema:\n      raise ValueError('RAW format does not take a schema')\n    field_names = [field.name for field in beam_schema.fields]\n    if len(field_names) != 1:\n      raise ValueError(f'Expecting exactly one field, found {field_names}')\n\n    def convert_to_bytes(row):\n      output = getattr(row, field_names[0])\n      if isinstance(output, bytes):\n        return output\n      elif isinstance(output, str):\n        return output.encode('utf-8')\n      else:\n        raise ValueError(\n            f\"Cannot encode payload for WriteToPubSub. \"\n            f\"Expected valid string or bytes object, \"\n            f\"got {repr(output)} of type {type(output)}.\")\n\n    return convert_to_bytes\n  elif format == 'JSON':\n    return json_utils.json_formater(beam_schema)\n  elif format == 'AVRO':\n    avro_schema = schema or avroio.beam_schema_to_avro_schema(beam_schema)\n    from_row = avroio.beam_row_to_avro_dict(avro_schema, beam_schema)\n\n    def formatter(row):\n      buffer = io.BytesIO()\n      fastavro.schemaless_writer(buffer, avro_schema, from_row(row))\n      buffer.seek(0)\n      return buffer.read()\n\n    return formatter","sourceCodeStart":278,"sourceCodeEnd":314,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/yaml/yaml_io.py#L278-L314","documentation":"The RAW-format converter for write_to_pubsub reads the single schema field from each row and expects bytes or str. If the value is any other type (int, dict, None, etc.), it raises this ValueError because a raw Pub/Sub payload must be bytes.","triggerScenarios":"Using format='RAW' where the sole field of the row holds a non-string/non-bytes value, e.g. an int or a parsed object.","commonSituations":"Pipeline emits numeric IDs or structured rows and the user assumes RAW will serialize them; RAW does no JSON/protobuf encoding.","solutions":["Switch format to JSON so any schema-encodable value is serialized","Coerce the field to str/bytes upstream (e.g. Map(lambda r: r._replace(payload=str(r.payload))))","Declare the field as STRING or BYTES in the schema and cast before writing"],"exampleFix":"// before\nbeam.Map(lambda row: row)  # payload is int\n// after\nbeam.Map(lambda row: {'payload': str(row.payload).encode('utf-8')})","handlingStrategy":"validation","validationCode":"val = getattr(row, field_name)\nif not isinstance(val, (bytes, str)):\n    raise TypeError(f'{field_name} must be bytes/str, got {type(val)}')","typeGuard":"def is_raw_encodable(v):\n    return isinstance(v, (bytes, str))","tryCatchPattern":"try:\n    pcoll | yaml_io.write_to_pubsub(...)\nexcept ValueError as e:\n    if 'Cannot encode payload' in str(e):\n        pcoll = pcoll | beam.Map(cast_payload_to_bytes)\n    else:\n        raise","preventionTips":["Declare the RAW field as STRING or BYTES in the schema","Cast values to bytes upstream before the sink","Prefer JSON when values are not raw bytes"],"tags":["python","apache-beam","pubsub","encoding","type-mismatch"],"backgroundTag":"type-mismatch","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}