Hello, I was looking for a Data Loader, that allow...
# hamilton-help
f
Hello, I was looking for a Data Loader, that allows to load a variable number of datasets based on a string template similar to the kedro dataset factories but couldn't find a direct equivalent in the documentation. Is there a suggested way to achieve this in Hamilton apart from defining the search path as one of the inputs of the dataflow apart from adjusting the examples given in the loading data documentation? I have seen that you can reuse a Kedro DataCatalog, but it doesn't look like the right approach to me to first setup some Kedro code just to pass it to Hamilton.
t
Hi! If I understood, this is about defining "templated" dataset references and creating multiple datasets at runtime. For example, Kedro uses this syntax to create multiple CSV datasets
Copy code
"{name}_data":
  type: pandas.CSVDataset
  filepath: data/01_raw/{name}_data.csv
## In Hamilton The materialization and with_materializers() pages are useful references. Personally, I would dynamically define
from_.csv(...)
data loaders and give them to the
Builder().with_materializer()
clause. Some example code
Copy code
import pathlib
from hamilton.driver import Builder
from hamilton.io.materialization import from_
import my_dataflow

loaders = []
for path in pathlib.Path("./data").iterdir():
   if path.is_file() and path.suffix == ".csv":
      csv_loader = from_.csv(target=f"{path.name}_data", path=path)
      loaders.append(csv_loader)

dr = (
  Builder()
  .with_modules(my_dataflow)
  .with_materializers(*loaders)
  .build()
)

# you should see the CSV loader nodes
dr.display_all_functions("dag.png")
Hamilton is less opinionated than Kedro when it comes to these things. Though, you should see how you can build your own "DataLoader builder" or catalog by creating utility functions
Copy code
def generate_csv_loaders(data_dir: str) -> list[DataLoader]:
   loaders = []
   for path in pathlib.Path(data_dir).iterdir():
     if path.is_file() and path.suffix == ".csv":
        csv_loader = from_.csv(target=f"{path.name}_data", path=path)
        loaders.append(csv_loader)
   return loaders
f
Thank you, this looks worth exploring. The approach I took yesterday looked something like this:
Copy code
def raw_data(
    *, path: Path, file_pattern: str
) -> dict[str, Any]:
    files = path.glob(file_pattern)
    collection: dict[str, Any] = {}
    for file in files:
        with open(file, "r") as f:
            collection[file.name] = json.load(f)
    return collection
I'll have a look into dynamically constructing the DataLoaders, though I'm wondering in how far this approach is going to turn out when dealing with hundred of files.
e
Yes what @Thierry Jean said!
Your approach is pretty good -- I don't think it'll have problems with hundreds of files.
One possiiblity is to define this using parallelizable/colletc
I'm not sure you can use the data loader framework for that but it'll help solve the problem