def _write(self, fo):
    buf = fastavro._writer.MemoryIO()
    block_writer = fastavro._writer.BLOCK_WRITERS[self._codec]
    n_records = 0
    n_block_records = 0

    # Cache a few variables.
    sync_interval = self._sync_interval
    write_data = fastavro._writer.write_data
    write_long = fastavro._writer.write_long

    def dump_header():
      """Write header."""
      metadata = {
        'avro.codec': self._codec,
        'avro.schema': dumps(self._schema),
      }
      fastavro._writer.write_header(fo, metadata, self._sync_marker)
      _logger.debug('Wrote header. Sync marker: %r', self._sync_marker)
      fastavro._writer.acquaint_schema(self._schema)

    def dump_data():
      """Write contents of memory buffer to file object."""
      write_long(fo, n_block_records)
      block_writer(fo, buf.getvalue())
      fo.write(self._sync_marker)
      buf.seek(0)
      buf.truncate(0)

    if self._schema:
      dump_header()
    try:
      while True:
        record = (yield)
        if not n_records:
          if not self._schema:
            self._schema = _infer_schema(record)
            _logger.info('Inferred schema: %s', dumps(self._schema))
            dump_header()
          schema = self._schema
        write_data(buf, record, schema)
        n_block_records += 1
        n_records += 1
        if buf.tell() >= sync_interval:
          dump_data()
          n_block_records = 0
    except GeneratorExit:
      if buf.tell():
        dump_data()
      fo.flush()
      _logger.info('Finished writing %s records.', n_records)
Coroutine to write to a file object.
dataset/ETHPy150Open mtth/hdfs/hdfs/ext/avro/__init__.py/AvroWriter._write