code: def get_mosaic_dataloader(
      mosaic_data_root,
      batch_size,
      num_physical_nodes,
      epoch_size=None,
      num_workers=None,
      cache_limit=None,
  ):
      # MosaicML StreamingDataset.
      use_s3 = mosaic_data_root.startswith("s3://")
  
      if not use_s3:
          assert (
              epoch_size is None
          ), "epoch_size not supported for streaming.LocalDataset"
          assert (
              cache_limit is None
          ), "cache_limit not supported for streaming.LocalDataset"
  
      if use_s3:
          MOSAIC_CACHE = "/tmp/mosaic_cache"
          try:
              import shutil
  
              shutil.rmtree(MOSAIC_CACHE)
          except (OSError, FileNotFoundError):
              pass
          streaming.base.util.clean_stale_shared_memory()
          print(
              f"Initializing mosaic StreamingDataset, cache_limit={cache_limit}"
          )
          mosaic_ds = S3MosaicDataset(
              s3_bucket=mosaic_data_root,
              num_physical_nodes=num_physical_nodes,
              cache_dir=MOSAIC_CACHE,
              cache_limit=cache_limit,
              epoch_size=epoch_size,
              transforms=get_transform(True),
          )
      else:
          mosaic_ds = MosaicDataset(
              mosaic_data_root, transforms=get_transform(True)
          )
  
      if num_workers is None:
          num_workers = os.cpu_count()
  
      print(
          f"Initializing torch DataLoader with {num_workers} workers."
      )
      mosaic_dl = torch.utils.data.DataLoader(
          mosaic_ds,
          batch_size=batch_size,
          num_workers=num_workers,
          drop_last=True,
      )
  
      return mosaic_dl
  
imports: import ray
  import torch
  import torchvision
  from torchvision.transforms.functional import pil_to_tensor
  import os
  import tensorflow
  import numpy
  from PIL import Image
  from typing import TYPE_CHECKING, Iterator, Callable, Any
  import pandas
  from benchmark import Benchmark
  import streaming
  from streaming import LocalDataset, StreamingDataset
  from ray.data._internal.delegating_block_builder import DelegatingBlockBuilder
label_desc: Use image node_id 11 for calling a global function with example usage: get_mosaic_dataloader(mosaic_data_root, batch_size, num_physical_nodes, epoch_size, num_workers, cache_limit) and returns: mosaic_dl