[ML-226] Migrate correlation daal to DPC++

examples/correlation/run-gpu-standalone.sh

@@ -9,7 +9,7 @@ DATA_FILE=$HDFS_ROOT/data/sample_kmeans_data.txt
 APP_JAR=target/oap-mllib-examples-$OAP_MLLIB_VERSION.jar
 APP_CLASS=org.apache.spark.examples.ml.CorrelationExample
 
-USE_GPU=true
+DEVICE=GPU
 RESOURCE_FILE=$PWD/IntelGpuResourceFile.json
 WORKER_GPU_AMOUNT=4
 EXECUTOR_GPU_AMOUNT=1
@@ -27,7 +27,7 @@ time $SPARK_HOME/bin/spark-submit --master $SPARK_MASTER \
     --conf "spark.sql.shuffle.partitions=$SPARK_DEFAULT_PARALLELISM" \
     --conf "spark.driver.extraClassPath=$SPARK_DRIVER_CLASSPATH" \
     --conf "spark.executor.extraClassPath=$SPARK_EXECUTOR_CLASSPATH" \
-    --conf "spark.oap.mllib.useGPU=$USE_GPU" \
+    --conf "spark.oap.mllib.device=$DEVICE" \
     --conf "spark.worker.resourcesFile=$RESOURCE_FILE" \
     --conf "spark.worker.resource.gpu.amount=$WORKER_GPU_AMOUNT" \
     --conf "spark.executor.resource.gpu.amount=$EXECUTOR_GPU_AMOUNT" \

examples/correlation/run.sh

@@ -2,7 +2,7 @@
 
 source ../../conf/env.sh
 
-
+DEVICE=CPU
 APP_JAR=target/oap-mllib-examples-$OAP_MLLIB_VERSION.jar
 APP_CLASS=org.apache.spark.examples.ml.CorrelationExample
 
@@ -17,6 +17,7 @@ time $SPARK_HOME/bin/spark-submit --master $SPARK_MASTER \
     --conf "spark.sql.shuffle.partitions=$SPARK_DEFAULT_PARALLELISM" \
     --conf "spark.driver.extraClassPath=$SPARK_DRIVER_CLASSPATH" \
     --conf "spark.executor.extraClassPath=$SPARK_EXECUTOR_CLASSPATH" \
+    --conf "spark.oap.mllib.device=$DEVICE" \
     --conf "spark.shuffle.reduceLocality.enabled=false" \
     --conf "spark.network.timeout=1200s" \
     --conf "spark.task.maxFailures=1" \

examples/pca/run.sh

@@ -6,6 +6,7 @@ APP_JAR=target/oap-mllib-examples-$OAP_MLLIB_VERSION.jar
 APP_CLASS=org.apache.spark.examples.ml.PCAExample
 
 # Dataset is created in the code, so no need to pass in as parameter
+DEVICE=CPU
 
 time $SPARK_HOME/bin/spark-submit --master $SPARK_MASTER \
     --num-executors $SPARK_NUM_EXECUTORS \
@@ -18,6 +19,7 @@ time $SPARK_HOME/bin/spark-submit --master $SPARK_MASTER \
     --conf "spark.sql.shuffle.partitions=$SPARK_DEFAULT_PARALLELISM" \
     --conf "spark.driver.extraClassPath=$SPARK_DRIVER_CLASSPATH" \
     --conf "spark.executor.extraClassPath=$SPARK_EXECUTOR_CLASSPATH" \
+    --conf "spark.oap.mllib.device=$DEVICE" \
     --conf "spark.shuffle.reduceLocality.enabled=false" \
     --conf "spark.network.timeout=1200s" \
     --conf "spark.task.maxFailures=1" \

mllib-dal/src/main/native/CorrelationOneAPIImpl.cpp

@@ -0,0 +1,90 @@
+/*******************************************************************************
+ * Copyright 2020 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 <chrono>
+
+#ifdef CPU_GPU_PROFILE
+#include "GPU.h"
+#ifndef ONEDAL_DATA_PARALLEL
+#define ONEDAL_DATA_PARALLEL
+#endif
+
+#include "Communicator.hpp"
+#include "OutputHelpers.hpp"
+#include "com_intel_oap_mllib_stat_CorrelationDALImpl.h"
+#include "oneapi/dal/algo/covariance.hpp"
+#include "oneapi/dal/table/homogen.hpp"
+#include "service.h"
+
+using namespace std;
+using namespace oneapi::dal;
+const int ccl_root = 0;
+
+static void doCorrelationOneAPICompute(JNIEnv *env, jint rankId,
+                                       jlong pNumTabData, jint executorNum,
+                                       const ccl::string &ipPort,
+                                       jint computeDeviceOrdinal,
+                                       jobject resultObj) {
+    std::cout << "oneDAL (native): GPU/CPU compute start , rankid = %ld "
+              << rankId << "; device = " << computeDeviceOrdinal
+              << "(0:HOST;1:CPU;2:GPU)" << std::endl;
+    const bool isRoot = (rankId == ccl_root);
+    ComputeDevice device = getComputeDeviceByOrdinal(computeDeviceOrdinal);
+    homogen_table htable =
+        *reinterpret_cast<const homogen_table *>(pNumTabData);
+
+    const auto cor_desc = covariance::descriptor{}.set_result_options(
+        covariance::result_options::cor_matrix |
+        covariance::result_options::means);
+    auto queue = getQueue(device);
+    auto comm = preview::spmd::make_communicator<preview::spmd::backend::ccl>(
+        queue, executorNum, rankId, ipPort);
+    const auto result_train = preview::compute(comm, cor_desc, htable);
+    if (isRoot) {
+        std::cout << "Mean:\n" << result_train.get_means() << std::endl;
+        std::cout << "Correlation:\n"
+                  << result_train.get_cor_matrix() << std::endl;
+        // Return all covariance & mean
+        jclass clazz = env->GetObjectClass(resultObj);
+
+        // Get Field references
+        jfieldID correlationNumericTableField =
+            env->GetFieldID(clazz, "correlationNumericTable", "J");
+
+        HomogenTablePtr correlation =
+            std::make_shared<homogen_table>(result_train.get_cor_matrix());
+        saveHomogenTablePtrToVector(correlation);
+
+        env->SetLongField(resultObj, correlationNumericTableField,
+                          (jlong)correlation.get());
+    }
+}
+
+JNIEXPORT jlong JNICALL
+Java_com_intel_oap_mllib_stat_CorrelationDALImpl_cCorrelationTrainDAL(
+    JNIEnv *env, jobject obj, jlong pNumTabData, jint executorNum,
+    jint computeDeviceOrdinal, jint rankId, jstring ipPort, jobject resultObj) {
+    std::cout << "oneDAL (native): use GPU DPC++ kernels with " << std::endl;
+    const char *ipPortPtr = env->GetStringUTFChars(ipPort, 0);
+    std::string ipPortStr = std::string(ipPortPtr);
+    printf("oneDAL (native):  CorrelationTrainDAL %d \n", computeDeviceOrdinal);
+    doCorrelationOneAPICompute(env, rankId, pNumTabData, executorNum, ipPortStr,
+                               computeDeviceOrdinal, resultObj);
+
+    env->ReleaseStringUTFChars(ipPort, ipPortPtr);
+    return 0;
+}
+#endif

mllib-dal/src/main/native/KMeansOneAPIImpl.cpp

@@ -17,7 +17,6 @@
 #include <chrono>
 #include <iomanip>
 #include <iostream>
-#include <mutex>
 
 #ifdef CPU_GPU_PROFILE
 #include "GPU.h"
@@ -42,7 +41,7 @@ static jlong doKMeansOneAPICompute(JNIEnv *env, jint rankId, jlong pNumTabData,
                                    jint computeDeviceOrdinal,
                                    jobject resultObj) {
     std::cout << "oneDAL (native): GPU/CPU compute start , rankid = " << rankId
-              << "; device = " << computeDeviceOrdinal << "(0:HOST;1:GPU;2:CPU)"
+              << "; device = " << computeDeviceOrdinal << "(0:HOST;1:CPU;2:GPU)"
               << std::endl;
     const bool isRoot = (rankId == ccl_root);
     ComputeDevice device = getComputeDeviceByOrdinal(computeDeviceOrdinal);

mllib-dal/src/main/native/Makefile

@@ -63,7 +63,7 @@ LIBS_COMMON := -L$(CCL_ROOT)/lib/cpu_gpu_dpcpp -lccl \
 
 ONEDAL_LIBS := -lonedal_dpc \
                -lonedal_core \
-               -lonedal_thread
+	       -lonedal_thread
 
 ifeq ($(PLATFORM_PROFILE),CPU_ONLY_PROFILE)
   LIBS := $(LIBS_COMMON) $(ONEDAL_LIBS)
@@ -78,7 +78,7 @@ CPP_SRCS += \
   ./ALSDALImpl.cpp ./ALSShuffle.cpp \
   ./NaiveBayesDALImpl.cpp \
   ./LinearRegressionDALImpl.cpp \
-  ./CorrelationDALImpl.cpp \
+  ./CorrelationOneAPIImpl.cpp \
   ./SummarizerDALImpl.cpp \
   ./oneapi/dal/HomogenTableImpl.cpp \
   ./oneapi/dal/SimpleMetadataImpl.cpp \
@@ -93,7 +93,7 @@ OBJS += \
   ./ALSDALImpl.o ./ALSShuffle.o \
   ./NaiveBayesDALImpl.o \
   ./LinearRegressionDALImpl.o \
-  ./CorrelationDALImpl.o \
+  ./CorrelationOneAPIImpl.o \
   ./SummarizerDALImpl.o \
   ./oneapi/dal/HomogenTableImpl.o \
   ./oneapi/dal/SimpleMetadataImpl.o \

mllib-dal/src/main/native/OutputHelpers.hpp

@@ -22,7 +22,7 @@
 #include "oneapi/dal/table/row_accessor.hpp"
 #include "oneapi/dal/table/common.hpp"
 
-std::ostream &operator<<(std::ostream &stream, const oneapi::dal::table &table) {
+inline std::ostream &operator<<(std::ostream &stream, const oneapi::dal::table &table) {
     auto arr = oneapi::dal::row_accessor<const float>(table).pull();
     const auto x = arr.get_data();