Skip to content

Allow Collection's update() and map() method to take dask arguments #28

Description

@robin-cls

When mapping a callback to the zcollection partitions, we might want to specify the 'resources' parameter when submitting the tasks.

zcollection.Collection.update:

    batches: Iterator[Sequence[str]] = dask_utils.split_sequence(
        selected_partitions, npartitions
        or dask_utils.dask_workers(client, cores_only=True))
    storage.execute_transaction(
        client, collection.synchronizer,
        # client.map(local_func, batches, resources=resources, key=func.__name__)) CURRENT CODE
        client.map(local_func, batches, resources=resources, key=func.__name__)) # REQUESTED CHANGE

In case we parallelize a long running task, dask worker may declare itself as idle if the blocking task releases the GIL or is blocked doing IO. Setting resources can help us limit overscheduling due to this behavior.

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions