Merge branch 'dev/eglaser-large-scale-incremental' into dev/eglaser-large-scale-incr

Author: ethanglaser

configs/README.md

@@ -117,6 +117,7 @@ Configs have the three highest parameter keys:
 |:---------------|:--------------|:--------|:------------|
 | `algorithm`:`estimator` | None |  | Name of measured estimator. |
 | `algorithm`:`estimator_params` | Empty `dict` |  | Parameters for estimator constructor. |
+| `algorithm`:`num_batches`:`training` | 5 |  | Number of batches to benchmark `partial_fit` function, using batches the size of number of samples specified (not samples divided by `num_batches`). For incremental estimators only. |
 | `algorithm`:`online_inference_mode` | False |  | Enables online mode for inference methods of estimator (separate call for each sample). |
 | `algorithm`:`sklearn_context` | None |  | Parameters for sklearn `config_context` used over estimator. |
 | `algorithm`:`sklearnex_context` | None |  | Parameters for sklearnex `config_context` used over estimator. Updated by `sklearn_context` if set. |

sklbench/benchmarks/sklearn_estimator.py

@@ -324,41 +324,33 @@ def verify_patching(stream: io.StringIO, function_name) -> bool:
     return acceleration_lines > 0 and fallback_lines == 0
 
 
-def create_online_function(
-    estimator_instance, method_instance, data_args, num_batches, batch_size
-):
+def create_online_function(estimator_instance, method_instance, data_args, num_batches):
 
     if "y" in list(inspect.signature(method_instance).parameters):
 
         def ndarray_function(x, y):
             for i in range(num_batches):
-                method_instance(
-                    x[i * batch_size : (i + 1) * batch_size],
-                    y[i * batch_size : (i + 1) * batch_size],
-                )
+                method_instance(x, y)
             if hasattr(estimator_instance, "_onedal_finalize_fit"):
                 estimator_instance._onedal_finalize_fit()
 
         def dataframe_function(x, y):
             for i in range(num_batches):
-                method_instance(
-                    x.iloc[i * batch_size : (i + 1) * batch_size],
-                    y.iloc[i * batch_size : (i + 1) * batch_size],
-                )
+                method_instance(x, y)
             if hasattr(estimator_instance, "_onedal_finalize_fit"):
                 estimator_instance._onedal_finalize_fit()
 
     else:
 
         def ndarray_function(x):
             for i in range(num_batches):
-                method_instance(x[i * batch_size : (i + 1) * batch_size])
+                method_instance(x)
             if hasattr(estimator_instance, "_onedal_finalize_fit"):
                 estimator_instance._onedal_finalize_fit()
 
         def dataframe_function(x):
             for i in range(num_batches):
-                method_instance(x.iloc[i * batch_size : (i + 1) * batch_size])
+                method_instance(x)
             if hasattr(estimator_instance, "_onedal_finalize_fit"):
                 estimator_instance._onedal_finalize_fit()
 
@@ -413,28 +405,20 @@ def measure_sklearn_estimator(
                         data_args = (x_train,)
                     else:
                         data_args = (x_test,)
+                batch_size = get_bench_case_value(
+                    bench_case, f"algorithm:batch_size:{stage}"
+                )
 
                 if method == "partial_fit":
-                    num_batches = get_bench_case_value(bench_case, "data:num_batches")
-                    batch_size = get_bench_case_value(bench_case, "data:batch_size")
-
-                    if batch_size is None:
-                        if num_batches is None:
-                            num_batches = 5
-                        batch_size = (
-                            data_args[0].shape[0] + num_batches - 1
-                        ) // num_batches
-                    if num_batches is None:
-                        num_batches = (
-                            data_args[0].shape[0] + batch_size - 1
-                        ) // batch_size
+                    num_batches = get_bench_case_value(
+                        bench_case, f"algorithm:num_batches:{stage}", 5
+                    )
 
                     method_instance = create_online_function(
                         estimator_instance,
                         method_instance,
                         data_args,
-                        num_batches,
-                        batch_size,
+                        num_batches
                     )
                 # daal4py model builders enabling branch
                 if enable_modelbuilders and stage == "inference":
@@ -452,6 +436,10 @@ def measure_sklearn_estimator(
                     metrics[method]["box filter mean[ms]"],
                     metrics[method]["box filter std[ms]"],
                 ) = measure_case(bench_case, method_instance, *data_args)
+                if batch_size is not None:
+                    metrics[method]["throughput[samples/ms]"] = (
+                        (data_args[0].shape[0] // batch_size) * batch_size
+                    ) / metrics[method]["time[ms]"]
                 if ensure_sklearnex_patching:
                     full_method_name = f"{estimator_class.__name__}.{method}"
                     sklearnex_logging_stream.seek(0)
@@ -559,9 +547,16 @@ def main(bench_case: BenchCase, filters: List[BenchCase]):
     for stage in estimator_methods.keys():
         data_descs[stage].update(
             {
-                "batch_size": get_bench_case_value(
-                    bench_case, f"algorithm:batch_size:{stage}"
-                )
+                key: val
+                for key, val in {
+                    "batch_size": get_bench_case_value(
+                        bench_case, f"algorithm:batch_size:{stage}"
+                    ),
+                    "num_batches": get_bench_case_value(
+                        bench_case, f"algorithm:num_batches:{stage}"
+                    )
+                }.items()
+                if val is not None
             }
         )
         if "n_classes" in data_description:

sklbench/report/implementation.py

@@ -97,6 +97,7 @@
     "order",
     "n_classes",
     "n_clusters",
+    "num_batches",
     "batch_size",
 ]
 
@@ -262,10 +263,7 @@ def get_summary_from_df(df: pd.DataFrame, df_name: str) -> pd.DataFrame:
             # only relative improvements are included in summary currently
             if len(column) > 1 and column[1] == f"{metric_name} relative improvement":
                 metric_columns.append(column)
-    if metric_columns:
-        summary = df[metric_columns].aggregate(geomean_wrapper, axis=0).to_frame().T
-    else:
-        summary = pd.DataFrame()
+    summary = df[metric_columns].aggregate(geomean_wrapper, axis=0).to_frame().T
     summary.index = pd.Index([df_name])
     return summary