make sklearn loader too - a lazy loader

Signed-off-by: dafnapension <[email protected]>

Author: dafnapension

performance/bluebench_profiler.py

@@ -76,7 +76,7 @@ def profiler_do_the_profiling(self, dataset_query: str, split: str, **kwargs):
                 benchmark_recipe=benchmark_recipe, split=split, **kwargs
             )
 
-            logger.critical(f"length of bluegench generated dataset: {len(dataset)}")
+            logger.critical(f"length of bluebench generated dataset: {len(dataset)}")
 
 
 dataset_query = "benchmarks.bluebench[loader_limit=30,max_samples_per_subset=30]"

src/unitxt/fusion.py

@@ -4,12 +4,9 @@
 from .dataclass import NonPositionalField
 from .operator import SourceOperator
 from .random_utils import new_random_generator
-from .settings_utils import get_settings
 from .stream import DynamicStream, MultiStream
 from .type_utils import isoftype
 
-settings = get_settings()
-
 
 class BaseFusion(SourceOperator):
     """BaseFusion operator that combines multiple multistreams into one.
@@ -37,11 +34,7 @@ def prepare_subsets(self):
             for i in range(len(self.subsets)):
                 self.named_subsets[i] = self.subsets[i]
         else:
-            for name, origin in self.subsets.items():
-                try:
-                    self.named_subsets[name] = origin
-                except Exception as e:
-                    raise RuntimeError(f"Exception in subset: {name}") from e
+            self.named_subsets = self.subsets
 
     def splits(self) -> List[str]:
         self.prepare_subsets()
@@ -78,30 +71,26 @@ def prepare(self):
 
     # flake8: noqa: C901
     def fusion_generator(self, split) -> Generator:
-        with settings.context(
-            disable_hf_datasets_cache=False,
-            allow_unverified_code=True,
-        ):
-            for origin_name, origin in self.named_subsets.items():
-                multi_stream = origin()
-                if split not in multi_stream:
-                    continue
-                emitted_from_this_split = 0
-                try:
-                    for instance in multi_stream[split]:
-                        if (
-                            self.max_instances_per_subset is not None
-                            and emitted_from_this_split >= self.max_instances_per_subset
-                        ):
-                            break
-                        if isinstance(origin_name, str):
-                            if "subset" not in instance:
-                                instance["subset"] = []
-                            instance["subset"].insert(0, origin_name)
-                        emitted_from_this_split += 1
-                        yield instance
-                except Exception as e:
-                    raise RuntimeError(f"Exception in subset: {origin_name}") from e
+        for origin_name, origin in self.named_subsets.items():
+            multi_stream = origin()
+            if split not in multi_stream:
+                continue
+            emitted_from_this_split = 0
+            try:
+                for instance in multi_stream[split]:
+                    if (
+                        self.max_instances_per_subset is not None
+                        and emitted_from_this_split >= self.max_instances_per_subset
+                    ):
+                        break
+                    if isinstance(origin_name, str):
+                        if "subset" not in instance:
+                            instance["subset"] = []
+                        instance["subset"].insert(0, origin_name)
+                    emitted_from_this_split += 1
+                    yield instance
+            except Exception as e:
+                raise RuntimeError(f"Exception in subset: {origin_name}") from e
 
 
 class WeightedFusion(BaseFusion):

src/unitxt/loaders.py

@@ -45,6 +45,7 @@
     Generator,
     Iterable,
     List,
+    Literal,
     Mapping,
     Optional,
     Sequence,
@@ -57,7 +58,6 @@
     IterableDataset,
     IterableDatasetDict,
     get_dataset_split_names,
-    load_dataset_builder,
 )
 from datasets import load_dataset as hf_load_dataset
 from huggingface_hub import HfApi
@@ -168,7 +168,7 @@ def load_data(self) -> MultiStream:
             self.__class__._loader_cache.max_size = settings.loader_cache_size
             self.__class__._loader_cache[str(self)] = iterables
         if isoftype(iterables, Dict[str, ReusableGenerator]):
-            return MultiStream.from_generators(iterables)
+            return MultiStream.from_generators(iterables, copying=True)
         return MultiStream.from_iterables(iterables, copying=True)
 
     def process(self) -> MultiStream:
@@ -259,9 +259,6 @@ def stream_dataset(self, split: str) -> Union[IterableDatasetDict, IterableDatas
                 )
             except ValueError as e:
                 if "trust_remote_code" in str(e):
-                    logger.critical(
-                        f"while raising trust_remote error, settings.allow_unverified_code = {settings.allow_unverified_code}"
-                    )
                     raise ValueError(
                         f"{self.__class__.__name__} cannot run remote code from huggingface without setting unitxt.settings.allow_unverified_code=True or by setting environment variable: UNITXT_ALLOW_UNVERIFIED_CODE."
                     ) from e
@@ -319,30 +316,28 @@ def load_iterables(
         if self.get_limit() is not None:
             self.log_limited_loading()
 
-        if not isinstance(self, LoadFromHFSpace):
-            # try the following for LoadHF only
-            if self.split is not None:
-                return {
-                    self.split: ReusableGenerator(
-                        self.split_generator, gen_kwargs={"split": self.split}
-                    )
-                }
+        if self.split is not None:
+            return {
+                self.split: ReusableGenerator(
+                    self.split_generator, gen_kwargs={"split": self.split}
+                )
+            }
 
-            try:
-                split_names = get_dataset_split_names(
-                    path=self.path,
-                    config_name=self.name,
-                    trust_remote_code=settings.allow_unverified_code,
+        try:
+            split_names = get_dataset_split_names(
+                path=self.path,
+                config_name=self.name,
+                trust_remote_code=settings.allow_unverified_code,
+            )
+            return {
+                split_name: ReusableGenerator(
+                    self.split_generator, gen_kwargs={"split": split_name}
                 )
-                return {
-                    split_name: ReusableGenerator(
-                        self.split_generator, gen_kwargs={"split": split_name}
-                    )
-                    for split_name in split_names
-                }
+                for split_name in split_names
+            }
 
-            except:
-                pass  # do nothing, and just continue to the usual load dataset
+        except:
+            pass  # do nothing, and just continue to the usual load dataset
             # self.split is None and
             # split names are not known before the splits themselves are loaded, and we need to load them here
 
@@ -495,14 +490,24 @@ def prepare(self):
         self.downloader = getattr(sklearn_datatasets, f"fetch_{self.dataset_name}")
 
     def load_iterables(self):
-        with TemporaryDirectory() as temp_directory:
-            for split in self.splits:
-                split_data = self.downloader(subset=split)
-                targets = [split_data["target_names"][t] for t in split_data["target"]]
-                df = pd.DataFrame([split_data["data"], targets]).T
-                df.columns = ["data", "target"]
-                df.to_csv(os.path.join(temp_directory, f"{split}.csv"), index=None)
-            return hf_load_dataset(temp_directory, streaming=False)
+        return {
+            split_name: ReusableGenerator(
+                self.split_generator, gen_kwargs={"split": split_name}
+            )
+            for split_name in self.splits
+        }
+
+    def split_generator(self, split: str) -> Generator:
+        dataset = self.__class__._loader_cache.get(str(self) + "_" + split, None)
+        if dataset is None:
+            split_data = self.downloader(subset=split)
+            targets = [split_data["target_names"][t] for t in split_data["target"]]
+            df = pd.DataFrame([split_data["data"], targets]).T
+            df.columns = ["data", "target"]
+            dataset = df.to_dict("records")
+            self.__class__._loader_cache.max_size = settings.loader_cache_size
+            self.__class__._loader_cache[str(self) + "_" + split] = dataset
+        yield from dataset
 
 
 class MissingKaggleCredentialsError(ValueError):

utils/.secrets.baseline

@@ -151,7 +151,7 @@
         "filename": "src/unitxt/loaders.py",
         "hashed_secret": "840268f77a57d5553add023cfa8a4d1535f49742",
         "is_verified": false,
-        "line_number": 565,
+        "line_number": 595,
         "is_secret": false
       }
     ],