apache/beam · error · ValueError

Cannot encode payload for WriteToPubSub. Expected valid stri

Error message

Cannot encode payload for WriteToPubSub. Expected valid string or bytes object, got {repr(output)} of type {type(output)}.

What it means

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.

Source

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

  if format.islower():
    format = format.upper()

  if format == 'RAW':
    if schema:
      raise ValueError('RAW format does not take a schema')
    field_names = [field.name for field in beam_schema.fields]
    if len(field_names) != 1:
      raise ValueError(f'Expecting exactly one field, found {field_names}')

    def convert_to_bytes(row):
      output = getattr(row, field_names[0])
      if isinstance(output, bytes):
        return output
      elif isinstance(output, str):
        return output.encode('utf-8')
      else:
        raise ValueError(
            f"Cannot encode payload for WriteToPubSub. "
            f"Expected valid string or bytes object, "
            f"got {repr(output)} of type {type(output)}.")

    return convert_to_bytes
  elif format == 'JSON':
    return json_utils.json_formater(beam_schema)
  elif format == 'AVRO':
    avro_schema = schema or avroio.beam_schema_to_avro_schema(beam_schema)
    from_row = avroio.beam_row_to_avro_dict(avro_schema, beam_schema)

    def formatter(row):
      buffer = io.BytesIO()
      fastavro.schemaless_writer(buffer, avro_schema, from_row(row))
      buffer.seek(0)
      return buffer.read()

    return formatter

View on GitHub (pinned to 12126d8942)

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

Example fix

// before
beam.Map(lambda row: row)  # payload is int
// after
beam.Map(lambda row: {'payload': str(row.payload).encode('utf-8')})

Defensive patterns

Strategy: validation

Validate before calling

val = getattr(row, field_name)
if not isinstance(val, (bytes, str)):
    raise TypeError(f'{field_name} must be bytes/str, got {type(val)}')

Type guard

def is_raw_encodable(v):
    return isinstance(v, (bytes, str))

Try / catch

try:
    pcoll | yaml_io.write_to_pubsub(...)
except ValueError as e:
    if 'Cannot encode payload' in str(e):
        pcoll = pcoll | beam.Map(cast_payload_to_bytes)
    else:
        raise

Prevention

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

When it happens

Trigger: 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.

Common situations: Pipeline emits numeric IDs or structured rows and the user assumes RAW will serialize them; RAW does no JSON/protobuf encoding.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

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