code: def shuffle_dataset(
      path_src: str,
      path_tgt: str,
      block_size: int = _DEFAULT_BLOCK_SIZE,
      bucket_size: int = _DEFAULT_SHUFFLE_BUCKET_SIZE,
      progress_bar: bool = False,
      output_name: Optional[str] = None,
  ):
      """Shuffle one distributed datataset, write results to another dataset.
  
      Args:
          path_str (str): path to source dataset
          path_tgt (str): path to write results
          block_size (int): dataset block size (default: 16MB)
          bucket_size (int): shuffle algorithm bucket size (default: 1GB)
          progress_bar (bool): show progress bar
  
      Example:
          >>> shuffle_dataset("/path/to/source", "/path/to/output")
      """
  
      if progress_bar and not support_tqdm:
          raise RuntimeError("Requires `tqdm` to enable progress bar.")
  
      ds = SimpleDataset(path_src, serializer=RawSerializer())
      num_buckets = (ds.nbytes + bucket_size - 1) // bucket_size
  
      tmp_files = [
          os.path.join(path_src, ".tmp.%s" % _random_string())
          for _ in range(num_buckets)
      ]
  
      try:
          # Step 1: write to bucket randomly
          f_tmp = [open(fname, "wb") for fname in tmp_files]
          try:
              iterator = ds
              if progress_bar:
                  iterator = tqdm(ds, desc="Shuffle step 1/2")
              for data in iterator:
                  bucket_id = int(random.random() * num_buckets)
                  len_data = len(data)
                  f_tmp[bucket_id].write(
                      struct.pack("I", len_data) + data
                  )
          finally:
              # close all files
              for fp in f_tmp:
                  if not fp.closed:
                      fp.close()
          f_tmp = []
  
          # Step 2: shuffle inside bucket
          if output_name is None:
              output_name = "%s.shuffle" % _random_string()
          with build_dataset(
              path_tgt,
              output_name,
              block_size=block_size,
              serializer=RawSerializer(),
          ) as writer:
              iterator = tmp_files
              if progress_bar:
                  iterator = tqdm(tmp_files, desc="Shuffle step 2/2")
  
              for fname in iterator:
                  fp = open(fname, "rb")
                  data_in_bucket = []
                  while True:
                      try:
                          raw_data = fp.read(4)
                          if len(raw_data) == 0:
                              # EOF
                              break
                          len_data = struct.unpack("I", raw_data)[0]
                          data_in_bucket.append(fp.read(len_data))
                      except EOFError:
                          break
                  random.shuffle(data_in_bucket)
                  for data in data_in_bucket:
                      writer.write(data)
                  fp.close()
                  os.unlink(fname)
      finally:
          # cleanup
          for fname in tmp_files:
              if os.path.exists(fname):
                  os.unlink(fname)
  
file_path: DeepKE/example/llm/CPM-Bee/src/cpm_live/dataset/utils.py
imports: import os
  import struct
  from typing import List, Optional
  from .distributed_dataset import SimpleDataset, build_dataset, _read_info_list, _write_info_list, _random_string, _DEFAULT_BLOCK_SIZE, FileInfo
  from .serializer import RawSerializer
  import random
  import shutil
label_desc: Use image node_id 1 for calling a global function with example usage: shuffle_dataset(path_src, path_tgt, block_size, bucket_size, progress_bar, output_name) without return types