Basic statistics algorithm interface in oneDAL (#1205)

Author: KulikovNikita

examples/sklearnex/spmd_basic_statistics.py

@@ -0,0 +1,61 @@
+# ===============================================================================
+# Copyright 2023 Intel Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ===============================================================================
+
+import numpy as np
+from mpi4py import MPI
+
+from dpctl import SyclQueue
+from sklearnex.spmd.basic_statistics import BasicStatistics as BasicStatisticsSpmd
+
+
+def generate_data(par, size, seed=777):
+    ns, nf = par['ns'], par['nf']
+
+    data_blocks, weight_blocks = [], []
+    rng = np.random.default_rng(seed)
+
+    for b in range(size):
+        data = rng.uniform(b, (b + 1) * (b + 1),
+                           size=(ns, nf))
+        weights = rng.uniform(1, (b + 1), size=ns)
+        weight_blocks.append(weights)
+        data_blocks.append(data)
+
+    data = np.concatenate(data_blocks, axis=0)
+    weights = np.concatenate(weight_blocks)
+
+    return (data, weights)
+
+
+q = SyclQueue("gpu")
+
+comm = MPI.COMM_WORLD
+rank = comm.Get_rank()
+size = comm.Get_size()
+
+params_spmd = {'ns': 19, 'nf': 31}
+
+data, weights = generate_data(params_spmd, size)
+weighted_data = np.diag(weights) @ data
+
+gtr_mean = np.mean(weighted_data, axis=0)
+gtr_std = np.std(weighted_data, axis=0)
+
+bss = BasicStatisticsSpmd(["mean", "standard_deviation"])
+res = bss.compute(data, weights, queue=q)
+
+print(f"Computed mean on rank {rank}:\n", res["mean"])
+print(f"Computed std on rank {rank}:\n", res["standard_deviation"])

examples/sklearnex/spmd_linear_regression.py

@@ -1,4 +1,4 @@
-#===============================================================================
+# ===============================================================================
 # Copyright 2023 Intel Corporation
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -12,54 +12,57 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-#===============================================================================
+# ===============================================================================
 
 import numpy as np
+from warnings import warn
+
 from mpi4py import MPI
-import dpctl
-from numpy.testing import assert_allclose
-from onedal.spmd.linear_model import LinearRegression as LinRegSpmd
+from dpctl import SyclQueue
+from sklearnex.spmd.linear_model import LinearRegression
 
 
-def generate_X_y(par, coef_seed, data_seed):
-    ns, nf, nr = par['ns'], par['nf'], par['nr']
+def generate_X_y(ns, data_seed):
+    nf, nr = 129, 131
 
-    crng = np.random.default_rng(coef_seed)
+    crng = np.random.default_rng(777)
     coef = crng.uniform(-4, 1, size=(nr, nf)).T
     intp = crng.uniform(-1, 9, size=(nr, ))
 
     drng = np.random.default_rng(data_seed)
     data = drng.uniform(-7, 7, size=(ns, nf))
     resp = data @ coef + intp[np.newaxis, :]
 
-    return data, resp, coef, intp
+    return data, resp
+
+
+def get_train_data(rank):
+    return generate_X_y(101, rank)
+
 
+def get_test_data(rank):
+    return generate_X_y(1024, rank)
 
-if __name__ == "__main__":
-    q = dpctl.SyclQueue("gpu")
 
-    comm = MPI.COMM_WORLD
-    rank = comm.Get_rank()
-    size = comm.Get_size()
+comm = MPI.COMM_WORLD
+rank = comm.Get_rank()
+size = comm.Get_size()
 
-    params_spmd = {'ns': 15, 'nf': 21, 'nr': 23}
-    params_grtr = {'ns': 77, 'nf': 21, 'nr': 23}
+if size < 2:
+    warn("This example was intentionally "
+         "designed to run in distributed mode only", RuntimeWarning)
 
-    Xsp, ysp, csp, isp = generate_X_y(params_spmd, size, size + rank - 1)
-    Xgt, ygt, cgt, igt = generate_X_y(params_grtr, size, size + rank + 1)
+X, y = get_train_data(rank)
 
-    assert_allclose(csp, cgt)
-    assert_allclose(isp, igt)
+queue = SyclQueue("gpu")
 
-    lrsp = LinRegSpmd(copy_X=True, fit_intercept=True)
-    lrsp.fit(Xsp, ysp, queue=q)
+model = LinearRegression().fit(X, y, queue)
 
-    assert_allclose(lrsp.coef_, csp.T)
-    assert_allclose(lrsp.intercept_, isp)
+print(f"Coefficients on rank {rank}:\n", model.coef_)
+print(f"Intercept on rank {rank}:\n", model.intercept_)
 
-    ypr = lrsp.predict(Xgt, queue=q)
+X_test, _ = get_test_data(rank)
 
-    assert_allclose(ypr, ygt)
+result = model.predict(X_test, queue)
 
-    print("Groundtruth responses:\n", ygt)
-    print("Computed responses:\n", ypr)
+print(f"Result on rank {rank}:\n", result)

onedal/__init__.py

@@ -46,7 +46,7 @@
     __all__.append('spmd')
 
 if daal_check_version((2023, 'P', 100)):
-    __all__.append('linear_model')
+    __all__ += ['basic_statistics', 'linear_model']
 
     if _is_dpc_backend:
-        __all__.append('spmd.linear_model')
+        __all__ += ['spmd.basic_statistics', 'spmd.linear_model']

onedal/basic_statistics/__init__.py

@@ -0,0 +1,19 @@
+#===============================================================================
+# Copyright 2023 Intel Corporation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#===============================================================================
+
+from .basic_statistics import BasicStatistics
+
+__all__ = ['BasicStatistics']

onedal/basic_statistics/basic_statistics.cpp

@@ -0,0 +1,174 @@
+/*******************************************************************************
+* Copyright 2023 Intel Corporation
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+*     http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*******************************************************************************/
+
+#include "oneapi/dal/algo/basic_statistics.hpp"
+
+#include "onedal/common.hpp"
+#include "onedal/version.hpp"
+
+#include <string>
+#include <regex>
+#include <map>
+
+namespace py = pybind11;
+
+namespace oneapi::dal::python {
+
+#if defined(ONEDAL_VERSION) && ONEDAL_VERSION >= 20230100
+
+namespace basic_statistics {
+
+template <typename Task, typename Ops>
+struct method2t {
+    method2t(const Task& task, const Ops& ops) : ops(ops) {}
+
+    template <typename Float>
+    auto operator()(const py::dict& params) {
+        using namespace dal::basic_statistics;
+
+        const auto method = params["method"].cast<std::string>();
+        ONEDAL_PARAM_DISPATCH_VALUE(method, "dense", ops, Float, method::dense);
+        ONEDAL_PARAM_DISPATCH_VALUE(method, "by_default", ops, Float, method::by_default);
+        ONEDAL_PARAM_DISPATCH_THROW_INVALID_VALUE(method);
+    }
+
+    Ops ops;
+};
+
+#define RESULT_OPTION(option) { #option, dal::basic_statistics::result_options::option }
+
+const std::map<std::string, dal::basic_statistics::result_option_id> result_option_registry {
+    RESULT_OPTION(min), RESULT_OPTION(max), RESULT_OPTION(sum), RESULT_OPTION(mean),
+    RESULT_OPTION(variance), RESULT_OPTION(variation), RESULT_OPTION(sum_squares),
+    RESULT_OPTION(standard_deviation), RESULT_OPTION(sum_squares_centered),
+    RESULT_OPTION(second_order_raw_moment)     
+};
+
+#undef RESULT_OPTION
+
+auto get_onedal_result_options(const py::dict& params) {
+    using namespace dal::basic_statistics;
+
+    auto result_option = params["result_option"].cast<std::string>();
+    result_option_id onedal_options;
+
+    try {
+        std::regex re("\\w+");
+        const std::sregex_iterator last{};
+        const std::sregex_iterator first( //
+            result_option.begin(),
+            result_option.end(),
+            re);
+
+        for (std::sregex_iterator it = first; it != last; ++it) {
+            const auto str = it->str();
+            const auto match = result_option_registry.find(str);
+            if (match == result_option_registry.cend()) {
+                ONEDAL_PARAM_DISPATCH_THROW_INVALID_VALUE(result_option);
+            } else {
+                onedal_options = onedal_options | match->second;
+            }
+        }
+    }
+    catch (std::regex_error& e) {
+        ONEDAL_PARAM_DISPATCH_THROW_INVALID_VALUE(result_option);
+    }
+
+    return onedal_options;
+}
+
+struct params2desc {
+    template <typename Float, typename Method, typename Task>
+    auto operator()(const py::dict& params) {
+        auto desc = dal::basic_statistics::descriptor<Float,
+            dal::basic_statistics::method::dense, dal::basic_statistics::task::compute>()
+            .set_result_options(get_onedal_result_options(params));
+        return desc;
+    }
+};
+
+template <typename Policy, typename Task>
+struct init_compute_ops_dispatcher {};
+
+template <typename Policy>
+struct init_compute_ops_dispatcher<Policy, dal::basic_statistics::task::compute> {
+    void operator()(py::module_& m) {
+        using Task = dal::basic_statistics::task::compute;
+        m.def("train",
+              [](const Policy& policy,
+                 const py::dict& params,
+                 const table& data,
+                 const table& weights) {
+                  using namespace dal::basic_statistics;
+                  using input_t = compute_input<Task>;
+
+                  compute_ops ops(policy, input_t{ data, weights }, params2desc{});
+                  return fptype2t{ method2t{ Task{}, ops } }(params);
+              });
+    }
+};
+
+template <typename Policy, typename Task>
+void init_compute_ops(py::module& m) {
+    init_compute_ops_dispatcher<Policy, Task>{}(m);
+}
+
+template <typename Task>
+void init_compute_result(py::module_& m) {
+    using namespace dal::basic_statistics;
+    using result_t = compute_result<Task>;
+
+    auto cls = py::class_<result_t>(m, "compute_result")
+                   .def(py::init())
+                   .DEF_ONEDAL_PY_PROPERTY(min, result_t)
+                   .DEF_ONEDAL_PY_PROPERTY(max, result_t)
+                   .DEF_ONEDAL_PY_PROPERTY(sum, result_t)
+                   .DEF_ONEDAL_PY_PROPERTY(mean, result_t)
+                   .DEF_ONEDAL_PY_PROPERTY(variance, result_t)
+                   .DEF_ONEDAL_PY_PROPERTY(variation, result_t)
+                   .DEF_ONEDAL_PY_PROPERTY(sum_squares, result_t)
+                   .DEF_ONEDAL_PY_PROPERTY(standard_deviation, result_t)
+                   .DEF_ONEDAL_PY_PROPERTY(sum_squares_centered, result_t)
+                   .DEF_ONEDAL_PY_PROPERTY(second_order_raw_moment, result_t);
+}
+
+ONEDAL_PY_DECLARE_INSTANTIATOR(init_compute_result);
+ONEDAL_PY_DECLARE_INSTANTIATOR(init_compute_ops);
+
+} // namespace basic_statistics
+
+ONEDAL_PY_INIT_MODULE(basic_statistics) {
+    using namespace dal::detail;
+    using namespace basic_statistics;
+    using namespace dal::basic_statistics;
+
+    auto sub = m.def_submodule("basic_statistics");
+    using task_list = types<dal::basic_statistics::task::compute>;
+
+#ifdef ONEDAL_DATA_PARALLEL_SPMD
+    ONEDAL_PY_INSTANTIATE(init_compute_ops, sub, policy_list_spmd, task_list);
+#else // ONEDAL_DATA_PARALLEL_SPMD
+    ONEDAL_PY_INSTANTIATE(init_compute_ops, sub, policy_list, task_list);
+#endif // ONEDAL_DATA_PARALLEL_SPMD
+
+    ONEDAL_PY_INSTANTIATE(init_compute_result, sub, task_list);
+}
+
+ONEDAL_PY_TYPE2STR(dal::basic_statistics::task::compute, "compute");
+
+#endif // defined(ONEDAL_VERSION) && ONEDAL_VERSION >= 20230100
+
+} // namespace oneapi::dal::python