kedro.io.PartitionedDataSet

PartitionedDataSet loads and saves partitioned file-like data using the underlying dataset definition. For filesystem level operations it uses fsspec: https://github.com/intake/filesystem_spec.

It also supports advanced features like lazy saving.

Example usage for the YAML API:

station_data:
  type: PartitionedDataSet
  path: data/03_primary/station_data
  dataset:
    type: pandas.CSVDataSet
    load_args:
      sep: '\t'
    save_args:
      sep: '\t'
      index: true
  filename_suffix: '.dat'

Example usage for the Python API:

>>> import pandas as pd
>>> from kedro.io import PartitionedDataSet
>>>
>>> # Create a fake pandas dataframe with 10 rows of data
>>> df = pd.DataFrame([{"DAY_OF_MONTH": str(i), "VALUE": i} for i in range(1, 11)])
>>>
>>> # Convert it to a dict of pd.DataFrame with DAY_OF_MONTH as the dict key
>>> dict_df = {
        day_of_month: df[df["DAY_OF_MONTH"] == day_of_month]
        for day_of_month in df["DAY_OF_MONTH"]
    }
>>>
>>> # Save it as small paritions with DAY_OF_MONTH as the partition key
>>> data_set = PartitionedDataSet(
        path="df_with_partition",
        dataset="pandas.CSVDataSet",
        filename_suffix=".csv"
    )
>>> # This will create a folder `df_with_partition` and save multiple files
>>> # with the dict key + filename_suffix as filename, i.e. 1.csv, 2.csv etc.
>>> data_set.save(dict_df)
>>>
>>> # This will create lazy load functions instead of loading data into memory immediately.
>>> loaded = data_set.load()
>>>
>>> # Load all the partitions
>>> for partition_id, partition_load_func in loaded.items():
        # The actual function that loads the data
        partition_data = partition_load_func()
>>>
>>> # Add the processing logic for individual partition HERE
>>> print(partition_data)

You can also load multiple partitions from a remote storage and combine them like this:

>>> import pandas as pd
>>> from kedro.io import PartitionedDataSet
>>>
>>> # these credentials will be passed to both 'fsspec.filesystem()' call
>>> # and the dataset initializer
>>> credentials = {"key1": "secret1", "key2": "secret2"}
>>>
>>> data_set = PartitionedDataSet(
        path="s3://bucket-name/path/to/folder",
        dataset="pandas.CSVDataSet",
        credentials=credentials
    )
>>> loaded = data_set.load()
>>> # assert isinstance(loaded, dict)
>>>
>>> combine_all = pd.DataFrame()
>>>
>>> for partition_id, partition_load_func in loaded.items():
        partition_data = partition_load_func()
        combine_all = pd.concat(
            [combine_all, partition_data], ignore_index=True, sort=True
        )
>>>
>>> new_data = pd.DataFrame({"new": [1, 2]})
>>> # creates "s3://bucket-name/path/to/folder/new/partition.csv"
>>> data_set.save({"new/partition.csv": new_data})