Use ray dataset and drop type casting in binary_feature prediction post processing for speedup

[magdyksaleh]

Code Pull Requests

Update the binary feature post-processing to use ray datasets to speed up overall time

Documentation Pull Requests

Note that the documentation HTML files are in docs/ while the Markdown sources are in mkdocs/docs.

If you are proposing a modification to the documentation you should change only the Markdown files.

api.md is automatically generated from the docstrings in the code, so if you want to change something in that file, first modify ludwig/api.py docstring, then run mkdocs/code_docs_autogen.py, which will create mkdocs/docs/api.md .

[github-actions]

Unit Test Results

       5 files  +       2         5 suites  +2   2h 4m 21s ⏱️ + 49m 15s
2 941 tests +       5  2 892 ✔️ +       5    49 💤 ±  0  0 ❌ ±0 
8 698 runs  +2 951  8 536 ✔️ +2 883  162 💤 +68  0 ❌ ±0 

Results for commit c06b7be. ± Comparison against base commit 06594a5.

♻️ This comment has been updated with latest results.

[tgaddair]

@magdyksaleh I'm going to suggest a different approach here, since I suspect the issue we're seeing here with binary_feature will be an issue for other output feature types as well.

The key idea is that all of the postprocess_predictions functions for all output features are only applying per-row transformations. There are not transformations being applied that require computing column-level metadata in this step. As such, instead of passing Dask DFs to the postprocess functions, we can run one top-level map_batches call that applies the transformations to batches as pandas dataframes.

Concretely:

1. Change the postprocess function in postprocessing.py to something like this:

def posproc_preds(df):
        for of_name, output_feature in output_features.items():
            return output_feature.postprocess_predictions(
                df,
                training_set_metadata[of_name],
            )
    
    predictions = backend.df_engine.map_batches(predictions, posproc_preds)

2. Change every postprocess_predictions function by removing output_directory and backend from the signature.

3. Change all the calls to backend.df_engine.map_objects in postprocess_predictions to simply call df.map, since it's just a Pandas DF.

[tgaddair]

I would rename to just map_batches, and maybe drop the batch_format param, since it would be difficult to make it work with other backends.

[magdyksaleh]

@magdyksaleh I'm going to suggest a different approach here, since I suspect the issue we're seeing here with binary_feature will be an issue for other output feature types as well.

The key idea is that all of the postprocess_predictions functions for all output features are only applying per-row transformations. There are not transformations being applied that require computing column-level metadata in this step. As such, instead of passing Dask DFs to the postprocess functions, we can run one top-level map_batches call that applies the transformations to batches as pandas dataframes.

Concretely:

1. Change the `postprocess` function in `postprocessing.py` to something like this:

def posproc_preds(df):
        for of_name, output_feature in output_features.items():
            return output_feature.postprocess_predictions(
                df,
                training_set_metadata[of_name],
            )
    
    predictions = backend.df_engine.map_batches(predictions, posproc_preds)

2. Change every `postprocess_predictions` function by removing `output_directory` and `backend` from the signature.

3. Change all the calls to `backend.df_engine.map_objects` in `postprocess_predictions` to simply call `df.map`, since it's just a Pandas DF.

I really like this suggestion! I'll try it out now. Thanks travis

[tgaddair]

Remove output_directory and backend

[tgaddair]

First element should be a dataframe, not a series.

[tgaddair]

LGTM! Please also double-check performance on higgs with this new approach before landing.

[magdyksaleh]

This is ready to merge pending CI passing