Allowing HyperbandSearchCV to also work with non dask arrays

[gioxc88]

I have a custom packages which uses a custom data-structure similar to pandas DataFrame but that also carry additional metadata. Let's call it CustomFrame.

I wrapped all estimator which share the sklearn API such as StandardScaler, Lasso, XGBRegressor etc.. to make them able to handle this CustomFrame.

This works well with dask implementation of GridSearchCV or RandomizedSearchCV as they just pass the X and y to the underlying estimators that, as I said, can correctly handle my CustomFrame.

I wanted to do the same with HyperbandSearchCV but I found this much more difficult, if not impossible without rewriting most of it, because the implementation is very different as it relies on Future objects and async await.

To be more precise at the moment it doesn't even work with pandas DataFrames. If you don't specify the test_size, the earliest time it fails is in dask_ml.model_selection._incremental at line 569 test_size = min(0.2, 1 / X.npartitions) because DataFrame does not have the attribute npartitions.

I would like to ask if you could consider adding the possibility to also handle also other ArrayLike objects the same way is already possible using GridSearchCV or RandomizedSearchCV

Many thanks
Gio

[stsievert]

I think specifying HyperbandSearchCV(..., test_size=0.2) will resolve this problem in the short term.

After that issue is circumvented, the method underlying HyperbandSearchCV works with Dask Dataframes, so if your CustomFrame inherits from dask.dataframe.Dataframe it should work:

dask-ml/dask_ml/model_selection/_incremental.py

Lines 181 to 182 in 7e11ce1

	if isinstance(X_train, (dd.DataFrame, dd.Series)):
	X_train = X_train.to_dask_array()

[stsievert]

What updates do you have on this @gioxc88? Did my solution work, or is there some other issue?

[gioxc88]

What updates do you have on this @gioxc88? Did my solution work, or is there some other issue?

Sorry for my late reply. I believe the problem is more complex than that and I didn't have time to test it because I have been working on something else for the past week but I will try to test it and get back to you.

Ideally I don't want to inherit from dask dataframe if possible. In general I believe that this hyperband search should have the same api as GridSearch. There is no need to force the user to use Dask Arrays

As far as the the problem with pandas dataframe I understand your solution would circumvent the problem, but shouldn't it be modify such that even pandas dataframe are accepted without errors (if you don't specify the test size of course)?

Any way I'll let you know asap, sorry for being slow, but I cannot focus on this 100% at the moment.

Gio

[gioxc88]

@stsievert
I am making some progress in order to make my CustomFrame work with hyperband.
I had to change how the train_test_split function works and now I am stuck at line 202 in
dask_ml.model_selection._incremental.py where it says

    # Convert to batches of delayed objects of numpy arrays
    X_train = sorted(futures_of(X_train), key=lambda f: f.key)
    y_train = sorted(futures_of(y_train), key=lambda f: f.key)

since so far I managed to do such that X_train, the argument of futures_of is a CustomFrame instance.

The problem is that the function futures_of returns now an empty list because looking into the source code my CustomFrame is not an instance of one of the following:

- (tuple, set, list)
- dict:
- SubgraphCallable
- Future

Consequently my code fails here at line 206 of the same module (dask_ml.model_selection._incremental.py):

train_eg = await client.gather(client.map(len, y_train))

with:

ValueError: min() arg is an empty sequence

Do you have any suggestion on how to overcome this without inheriting from dask Dataframe?

Many thanks.
Gio

[stsievert]

same API as Scikit-learn GridSearchCV

I see your point now. It'd be nice to support this. I think anything larger than memory will be difficult.

tips

I think that line with "train_eg" is only used for info printing ("X has between X and Y examples per chunk"). It's not critical. It might be enough to add an if-statement:

train_eg = client.map(len, y_train)
if len(train_eg):
    msg = "...".format(min(train_eg))
    logger.info(msg)

I think this would be a good PR.

That said, I'd probably add a __len__ function to your CustomFrame.

[gioxc88]

same API as Scikit-learn GridSearchCV

I see your point now. It'd be nice to support this. I think anything larger than memory will be difficult.

tips

I think that line with "train_eg" is only used for info printing ("X has between X and Y examples per chunk"). It's not critical. It might be enough to add an if-statement:

train_eg = client.map(len, y_train)
if len(train_eg):
    msg = "...".format(
        min(train_eg)
    )
    logger.info(msg)

I think this would be a good PR.

That said, I'd probably add a __len__ function to your CustomFrame.

Thank you, I accidentally closed the issue.
The CustomFrame already has __len__ implemented, but this is not what is causing the error there.

Anyway the solution can just be that the custom frame could be something like

npartitions = getattr(X, 'npartitions', None)
if npartitions:
    # use the current implementation
else: # meaning that X does not support larger than memory
    # use a simpler implementation which does not require futures object

at the moment this is what I am trying to implement.

[gioxc88]

@stsievert I have a found a solution that will allow Hyperband to work with other Array-like objects.
Would you consider a PR to make it available to anyone?

The only change is related to the dask_ml.model_selection._incremental.py module at line 201:

Current implementation:

# Convert to batches of delayed objects of numpy arrays
X_train = sorted(futures_of(X_train), key=lambda f: f.key)
y_train = sorted(futures_of(y_train), key=lambda f: f.key)
assert len(X_train) == len(y_train)

train_eg = await client.gather(client.map(len, y_train))
msg = "[CV%s] For training there are between %d and %d examples in each chunk"
logger.info(msg, prefix, min(train_eg), max(train_eg))

# Order by which we process training data futures
order = []

def get_futures(partial_fit_calls):
    """Policy to get training data futures

    Currently we compute once, and then keep in memory.
    Presumably in the future we'll want to let data drop and recompute.
    This function handles that policy internally, and also controls random
    access to training data.
    """
    # Shuffle blocks going forward to get uniform-but-random access
    while partial_fit_calls >= len(order):
        L = list(range(len(X_train)))
        rng.shuffle(L)
        order.extend(L)
    j = order[partial_fit_calls]
    return X_train[j], y_train[j]

Proposed implementation

# Order by which we process training data futures
order = []

if hasattr(X_train, 'npartitions'):
    # Convert to batches of delayed objects of numpy arrays
    X_train = sorted(futures_of(X_train), key=lambda f: f.key)
    y_train = sorted(futures_of(y_train), key=lambda f: f.key)
    assert len(X_train) == len(y_train)

    train_eg = await client.gather(client.map(len, y_train))
    msg = "[CV%s] For training there are between %d and %d examples in each chunk"
    logger.info(msg, prefix, min(train_eg), max(train_eg))

    def get_futures(partial_fit_calls):
        """Policy to get training data futures
        Currently we compute once, and then keep in memory.
        Presumably in the future we'll want to let data drop and recompute.
        This function handles that policy internally, and also controls random
        access to training data.
        """
        # Shuffle blocks going forward to get uniform-but-random access
        while partial_fit_calls >= len(order):
            L = list(range(len(X_train)))
            rng.shuffle(L)
            order.extend(L)
        j = order[partial_fit_calls]
        return X_train[j], y_train[j]

else:
    def get_futures(partial_fit_calls):
        return X_train, y_train

I only put the current code in an if block and changed the policy to simply return X_train and y_train if the Array-like object has no attribute npartitions.

What do you think?
Gio

[gioxc88]

PR: #751