apache/beam · error · ValueError

compressor not initialized

Error message

compressor not initialized

What it means

CompressedFile lazily creates its compressor only when opened for writing; write() requires that compressor to exist. Calling write on a CompressedFile that was constructed (or reopened) in read mode leaves self._compressor as None, so Beam raises ValueError.

Solutions

1. Open the underlying file (or FileSystems destination) in write mode ('wb'/'ab') so the compressor is initialized.
2. Check writable() (or mode) before writing and branch to a reader instead.
3. Fix the code path that hands a read-mode handle to a write API.

Example fix

# before
fh = open(path, 'rb')
out = CompressedFile(fh, compression_type=CompressionTypes.GZIP)
out.write(b'data')  # raises
# after
fh = open(path, 'wb')
out = CompressedFile(fh, compression_type=CompressionTypes.GZIP)
out.write(b'data')

if not compressed_file.writable():
    raise ValueError('CompressedFile opened in read mode; cannot write')

def can_write(f) -> bool:
    return hasattr(f, 'writable') and f.writable()

try:
    out.write(data)
except ValueError as e:
    if 'compressor not initialized' in str(e):
        raise IOError('attempted write on read-mode CompressedFile') from e
    raise

When it happens

Trigger: Opening a file in read mode (or via a path where writable() is False), then calling write(); using a CompressedFile after a mode reset that cleared the compressor.

Common situations: Opening with 'rb' but a code path still calls write(); a sink writing to a read-only file handle; accidentally swapping reader/writer objects.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• decompressor not initialized
• A schema is required to write non-schema'd data.
• A sink must inherit iobase.Sink, iobase.NativeSink, or be a…
• An explicit schema is required to write non-schema'd…
• Cannot call read after iterating.

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

Appendix: source

Thrown at sdks/python/apache_beam/io/filesystem.py:229

    elif self._compression_type == CompressionTypes.LZMA:
      self._compressor = lzma.LZMACompressor()
    else:
      assert self._compression_type == CompressionTypes.GZIP
      self._compressor = zlib.compressobj(
          zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, self._gzip_mask)

  def readable(self) -> bool:
    mode = self._file.mode
    return 'r' in mode or 'a' in mode

  def writeable(self) -> bool:
    mode = self._file.mode
    return 'w' in mode or 'a' in mode

  def write(self, data: bytes) -> None:
    """Write data to file."""
    if not self._compressor:
      raise ValueError('compressor not initialized')
    self._uncompressed_position += len(data)
    compressed = self._compressor.compress(data)
    if compressed:
      self._file.write(compressed)

  def _fetch_to_internal_buffer(self, num_bytes: int) -> None:
    """Fetch up to num_bytes into the internal buffer."""
    if (not self._read_eof and self._read_position > 0 and
        (self._read_buffer.tell() - self._read_position) < num_bytes):
      # There aren't enough number of bytes to accommodate a read, so we
      # prepare for a possibly large read by clearing up all internal buffers
      # but without dropping any previous held data.
      self._read_buffer.seek(self._read_position)
      data = self._read_buffer.read()
      self._clear_read_buffer()
      self._read_buffer.write(data)

    assert self._decompressor

View on GitHub (pinned to 12126d8942)