DatasetError: Failed while saving data to data set PickleDataset(backend=pickle,
filepath=.../performance_optimisation/client/data/08_reporting/explaine
r.joblib, load_args=
{}
, protocol=s3, save_args=
{}
).
[Errno 22] Part number must be an integer between 1 and 10000, inclusive

ParittionedDataSet

import pickle
import fsspec

# Function to write data in chunks
def write_pickle_in_chunks(data, file_path, chunk_size=1024):
    pickled_data = pickle.dumps(data)
    with fsspec.open(file_path, 'wb') as f:
        for i in range(0, len(pickled_data), chunk_size):
            chunk = pickled_data[i:i+chunk_size]
            f.write(chunk)

# Function to read data in chunks and reconstruct the pickle
def read_pickle_in_chunks(file_path, chunk_size=1024):
    chunks = []
    with fsspec.open(file_path, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            chunks.append(chunk)
    pickled_data = b''.join(chunks)
    data = pickle.loads(pickled_data)
    return data

# Sample data to pickle
data = {'key': 'value', 'number': 42, 'list': [1, 2, 3, 4, 5]}

# Specify the file path
file_path = 'data.pickle'

# Write data in chunks
write_pickle_in_chunks(data, file_path)

# Read data in chunks and reconstruct the original data
reconstructed_data = read_pickle_in_chunks(file_path)

# Verify that the reconstructed data matches the original data
print(reconstructed_data == data)  # Should print: True

import boto3
import joblib
import io
from kedro.io.core import AbstractDataset, DatasetError

class S3MultipartPickleDataSet(AbstractDataset):
    def __init__(self, filepath: str, save_args: dict = None):
        self._filepath = filepath
        self._save_args = save_args or {}
        self._s3 = boto3.client('s3')

    def _describe(self) -> dict:
        return dict(filepath=self._filepath, save_args=self._save_args)

    def _load(self) -> object:
        raise DatasetError("Load not implemented")

    def _save(self, data: object) -> None:
        try:
            # Create an in-memory bytes buffer
            with io.BytesIO() as temp_buffer:
                # Dump the data to the in-memory buffer using joblib
                joblib.dump(data, temp_buffer)
                # Ensure the buffer's cursor is at the start
                temp_buffer.seek(0)
                # Stream the buffer to S3
                bucket, key = self._parse_s3_url(self._filepath)
                self._s3.upload_fileobj(temp_buffer, bucket, key, Config=self._get_transfer_config())
        except Exception as exc:
            raise DatasetError(f"Failed to save data to {self._filepath}") from exc

    def _get_transfer_config(self):
        return boto3.s3.transfer.TransferConfig(
            multipart_threshold=int(self._save_args.get('multipart_chunksize', 25 * 1024 * 1024)),
            max_concurrency=10,
            multipart_chunksize=int(self._save_args.get('multipart_chunksize', 25 * 1024 * 1024)),
            use_threads=True
        )

    @staticmethod
    def _parse_s3_url(s3_url: str):
        assert s3_url.startswith('s3://')
        bucket_key = s3_url[len('s3://'):].split('/', 1)
        return bucket_key[0], bucket_key[1]

PartitionedPickleDataset

< its right in the name:p

kedro-datasets