[Re-opened elsewhere] Handle nullable types and empty partitions before Dask-ML predict

[sarahyurick]

Tagging @randerzander and @VibhuJawa

Changes in create_model.py handle nullable types, such as for this example:

df = pd.DataFrame({
    "rough_day_of_year": pd.Series([0, 1, 2, 3], dtype='Int32'),
    "prev_day_inches_rained": pd.Series([0, 1, 2, 3], dtype='float32'),
    "rained": pd.Series([False, False, False, True])
})
c.create_table("train_set", df)

model_class = ".linear_model.LogisticRegression'"
if GPU:
    model_class = "'cuml" + model_class
else:
    model_class = "'sklearn" + model_class

c.sql(f"""
CREATE OR REPLACE MODEL model WITH (
    model_class = {model_class},
     wrap_predict = True,
     wrap_fit = False,
    target_column = 'rained'
) AS (
    SELECT *
    FROM train_set
)
""")

c.sql("""
SELECT * FROM PREDICT(
  MODEL model,
  SELECT * FROM train_set
)
""").compute()

Changes in predict.py handle empty partitions, modeled based on this Dask-ML PR: dask/dask-ml#912

[sarahyurick]

With this, output_meta is always []. Should this maybe be in some sort of try/except block since we're only handling the CPU case?

[VibhuJawa]

I dont think we can just hardcode the meta to be output_meta to be np.array([]) . We also use cuML for this case and that outputs a cuDF Series.

[sarahyurick]

compute() is needed on the Delayed object to get the number of rows in the partition. I believe that right now, output_meta will always be []?

[VibhuJawa]

You dont need to compute for this to do this, we can do it lazily too.

[sarahyurick]

AttributeError: 'KMeans' object has no attribute 'predict_meta'

[VibhuJawa]

We should not hard code any meta values and should only handle cases when model is ParallelPostFit .

[VibhuJawa]

I dont think we can just hardcode the meta to be output_meta to be np.array([]) . We also use cuML for this case and that outputs a cuDF Series.

[VibhuJawa]

You dont need to compute for this to do this, we can do it lazily too.

[VibhuJawa]

We should wrap the predict like the following for cases only for when we have a ParallelPostFit model.

  if isinstance(model, ParallelPostFit):
      output_meta = model.predict_meta
      if predict_meta is None:
         predict_meta = model.estimator.predict(part._meta_nonempty)
         
      prediction  = part.map_partitions(_predict, predict_meta, model.estimator, meta=predict_meta)

def _pedict(part, predict_meta, estimator):
    if part.shape[0] == 0 and predict_meta is not None:
        empty_output = handle_empty_partitions(output_meta)
        return empty_output
    return estimator.predict(part)

[VibhuJawa]

Please add more tests