kedro.io.PartitionedDataSet¶

class kedro.io.PartitionedDataSet(path, dataset, filepath_arg='filepath', filename_suffix='', credentials=None, load_args=None)[source]¶

Bases: kedro.io.core.AbstractDataSet

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.

Example:

import pandas as pd
from kedro.io import PartitionedDataSet

credentials = {
    "key1": "secret1",  # will be passed to 'fsspec.filesystem()' call
    "dataset_credentials": {  # will be passed to the dataset initializer
        "key2": "secret2",
        "key3": "secret3"
    }
}

data_set = PartitionedDataSet(
    path="s3://bucket-name/path/to/folder",
    dataset="CSVS3DataSet",
    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})

Methods

`PartitionedDataSet.__init__`(path, dataset[, …])	Creates a new instance of `PartitionedDataSet`.
`PartitionedDataSet.exists`()	Checks whether a data set’s output already exists by calling the provided _exists() method.
`PartitionedDataSet.from_config`(name, config)	Create a data set instance using the configuration provided.
`PartitionedDataSet.get_last_load_version`()	Versioned datasets should override this property to return last loaded version
`PartitionedDataSet.get_last_save_version`()	Versioned datasets should override this property to return last saved version.
`PartitionedDataSet.invalidate_cache`()	Invalidate _list_partitions method and underlying filesystem caches.
`PartitionedDataSet.load`()	Loads data by delegation to the provided load method.
`PartitionedDataSet.release`()	Release any cached data.
`PartitionedDataSet.save`(data)	Saves data by delegation to the provided save method.

__init__(path, dataset, filepath_arg='filepath', filename_suffix='', credentials=None, load_args=None)[source]¶

Creates a new instance of PartitionedDataSet.

Parameters:

path (str) – Path to the folder containing partitioned data. If path starts with the protocol (e.g., s3://) then the corresponding fsspec concrete filesystem implementation will be used. If protocol is not specified, fsspec.implementations.local.LocalFileSystem will be used. Note: Some concrete implementations are bundled with fsspec, while others (like s3 or gcs) must be installed separately prior to usage of the PartitionedDataSet.
dataset (Union[str, Type[AbstractDataSet], Dict[str, Any]]) – Underlying dataset definition. This is used to instantiate the dataset for each file located inside the path. Accepted formats are: a) object of a class that inherits from AbstractDataSet b) a string representing a fully qualified class name to such class c) a dictionary with type key pointing to a string from b), other keys are passed to the Dataset initializer. Note: Credentials resolution is not currently supported for the underlying dataset definition.
filepath_arg (str) – Underlying dataset initializer argument that will contain a path to each corresponding partition file. If unspecified, defaults to “filepath”.
filename_suffix (str) – If specified, only partitions that end with this string will be processed.
credentials (Optional[Dict[str, Any]]) –
Protocol-specific options that will be passed to fsspec.filesystem call: https://filesystem-spec.readthedocs.io/en/latest/api.html#fsspec.filesystem _and_ also to the underlying dataset initializer. If dataset_credentials key is present in this dictionary, then only its value will be passed to the dataset initializer credentials argument instead of the copy of the entire dictionary.

Example 1: If credentials = {"k1": "secret1"}, then filesystem

is called as filesystem(..., k1="secret1"), the dataset is instantiated as dataset_class(..., credentials={"k1": "secret1"}).

Example 2: If

credentials = {"k1": "secret1", "dataset_credentials": {"k2": "secret2"}}, then filesystem is called as filesystem(..., k1="secret1"), the dataset is instantiated as dataset_class(..., credentials={"k2": "secret2"}).

Example 3: If

credentials = {"dataset_credentials": {"k2": "secret2"}}, then credentials are not passed to the filesystem call, the dataset is instantiated as dataset_class(..., credentials={"k2": "secret2"}).

Example 4: If

credentials = {"k1": "secret1", "dataset_credentials": None}, then filesystem is called as filesystem(..., k1="secret1"), credentials are not passed to the dataset initializer.
load_args (Optional[Dict[str, Any]]) – Keyword arguments to be passed into find() method of the filesystem implementation.

Raises:

DataSetError – If versioning is enabled for the underlying dataset.

exists()¶

Checks whether a data set’s output already exists by calling the provided _exists() method.

Return type:	`bool`
Returns:	Flag indicating whether the output already exists.
Raises:	`DataSetError` – when underlying exists method raises error.

classmethod from_config(name, config, load_version=None, save_version=None)¶

Create a data set instance using the configuration provided.

Parameters:	name (`str`) – Data set name. config (`Dict`[`str`, `Any`]) – Data set config dictionary. load_version (`Optional`[`str`]) – Version string to be used for `load` operation if the data set is versioned. Has no effect on the data set if versioning was not enabled. save_version (`Optional`[`str`]) – Version string to be used for `save` operation if the data set is versioned. Has no effect on the data set if versioning was not enabled.
Return type:	`AbstractDataSet`
Returns:	An instance of an `AbstractDataSet` subclass.
Raises:	`DataSetError` – When the function fails to create the data set from its config.

get_last_load_version()¶

Versioned datasets should override this property to return last loaded version

Return type:	`Optional`[`str`]

get_last_save_version()¶

Versioned datasets should override this property to return last saved version.

Return type:	`Optional`[`str`]

invalidate_cache()[source]¶: Invalidate _list_partitions method and underlying filesystem caches.

load()¶

Loads data by delegation to the provided load method.

Return type:	`Any`
Returns:	Data returned by the provided load method.
Raises:	`DataSetError` – When underlying load method raises error.

release()¶

Release any cached data.

Raises:	`DataSetError` – when underlying exists method raises error.
Return type:	`None`

save(data)¶

Saves data by delegation to the provided save method.

Parameters:	data (`Any`) – the value to be saved by provided save method.
Raises:	`DataSetError` – when underlying save method raises error.
Return type:	`None`