num_examples_progress_update = 0
        writer = ArrowWriter(
            features=shard.features,
            path=fpath,
            storage_options=storage_options,
            embed_local_files=True,
        )
        try:
            _time = time.time()
            for pa_table in shard.with_format("arrow").iter(batch_size):
                writer.write_table(pa_table)
                num_examples_progress_update += len(pa_table)
                if time.time() > _time + config.PBAR_REFRESH_TIME_INTERVAL:
                    _time = time.time()
                    yield job_id, False, num_examples_progress_update
                    num_examples_progress_update = 0
        finally:
            yield job_id, False, num_examples_progress_update
            num_examples, num_bytes = writer.finalize()
            writer.close()

        yield job_id, True, (num_examples, num_bytes)