Help needed: How to filter a remote directory for IncrementalDataSet

[drdebian]

I'm trying to pull in data from a remote SFTP directory into an IncrementalDataSet. There are groups of files like car1*.csv, car2*.csv, ..., somethingelse*.csv in the same directory.

I found no obvious way to use wildcards to filter those groups of files. What I would like to have is something like this:

my_incremental_dataset:
  type: IncrementalDataSet
  path: sftp:///path/to/raw/data/car8*
  filename_suffix: .csv
  credentials: my_credentials
  dataset: pandas.CSVDataSet

My expectation would be, that I get all files from remote directory /path/to/raw/data/ matching the pattern of "car8*.csv" read into my_incremental_dataset. Writing it like above, however, gets me no files at all. If I leave away the *, I get all files mangled into one dataset, which does not work due to different schemas of those file groups.

Any help would be greatly appreciated!

[datajoely]

Hi @drdebian I'm actually not support this out of the box. We use a library called fsspec under the hood which does support 'glob' paths.

If you look at the implementation for CSVDataSet we do pass fs_args directly to thefsspec.filesystem() constructor which may allow this to work.

I would try passing different arguments to fsspec or perhaps implement your own custom CSVDataSet by inheriting and overriding our own.

[drdebian]

Hi @drdebian I'm actually not support this out of the box. We use a library called fsspec under the hood which does support 'glob' paths.
If you look at the implementation for CSVDataSet we do pass fs_args directly to thefsspec.filesystem() constructor which may allow this to work.

Thanks for the good idea. I explored this option by adding fs_args to the catalog.yml:

my_incremental_dataset:
  type: IncrementalDataSet
  path: sftp:///path/to/my/incr/data/
  filename_suffix: .csv
  credentials: my_credentials
  dataset: 
    type: pandas.CSVDataSet
    fs_args: 
      glob: '*car8*'

Unfortunately, the fs_args seem to end up in paramiko's sftp._connect() method and thus won't work, because it knows nothing about "glob":

Traceback (most recent call last):
  File "C:\Users\Albert\anaconda3\envs\py38\lib\site-packages\kedro\io\core.py", line 182, in load
    return self._load()
  File "C:\Users\Albert\anaconda3\envs\py38\lib\site-packages\kedro\io\partitioned_data_set.py", line 499, in _load
    partitions[partition_id] = self._dataset_type(  # type: ignore
  File "C:\Users\Albert\anaconda3\envs\py38\lib\site-packages\kedro\extras\datasets\pandas\csv_dataset.py", line 120, in __init__
    self._fs = fsspec.filesystem(self._protocol, **_credentials, **_fs_args)
  File "C:\Users\Albert\anaconda3\envs\py38\lib\site-packages\fsspec\registry.py", line 253, in filesystem
    return cls(**storage_options)
  File "C:\Users\Albert\anaconda3\envs\py38\lib\site-packages\fsspec\spec.py", line 76, in __call__
    obj = super().__call__(*args, **kwargs)
  File "C:\Users\Albert\anaconda3\envs\py38\lib\site-packages\fsspec\implementations\sftp.py", line 47, in __init__
    self._connect()
  File "C:\Users\Albert\anaconda3\envs\py38\lib\site-packages\fsspec\implementations\sftp.py", line 53, in _connect
    self.client.connect(self.host, **self.ssh_kwargs)
TypeError: **connect() got an unexpected keyword argument 'glob'**

Looking at the implementation of filename_suffix in kedro/io/partitioned_data_set.py, I think a possible implementation would probably have to involve some modification to this part of the the code:

So I think all that's really missing is the concept of a filename_prefix (in my case 'car8*') which is used to filter the partitions in the same way it is already being done for the filename_suffix.

[drdebian]

Hi @drdebian
I would try passing different arguments to fsspec or perhaps implement your own custom CSVDataSet by inheriting and overriding our own.

I actually went ahead and added the missing filename_prefix feature to the codebase. You can find the pull request here: #1211