[ML-254] [Random forest] Enable Random Forest Regressor algorithm on OAP-Mllib

examples/random-forest-pyspark/random_forest_regressor_example.py

@@ -0,0 +1,75 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+#
+
+"""
+Random Forest Regressor Example.
+"""
+from pyspark.ml import Pipeline
+from pyspark.ml.regression import RandomForestRegressor
+from pyspark.ml.feature import VectorIndexer
+from pyspark.ml.evaluation import RegressionEvaluator
+from pyspark.sql import SparkSession
+
+if __name__ == "__main__":
+    spark = SparkSession \
+        .builder \
+        .appName("RandomForestRegressorExample") \
+        .getOrCreate()
+
+    if (len(sys.argv) != 2) :
+        print("Require data file path as input parameter")
+        sys.exit(1)
+
+    # Load and parse the data file, converting it to a DataFrame.
+    data_sparse = spark.read.format("libsvm").load(sys.argv[1]).toDF("label", "features_sparse")
+    data = data_sparse.rdd.map(lambda x: Row(label=x[0], features=DenseVector(x[1].toArray()))).toDF()
+    data.printSchema()
+    data.show()
+
+    # Automatically identify categorical features, and index them.
+    # Set maxCategories so features with > 4 distinct values are treated as continuous.
+    featureIndexer = \
+        VectorIndexer(inputCol="features", outputCol="indexedFeatures", maxCategories=4).fit(data)
+
+    # Split the data into training and test sets (30% held out for testing)
+    (trainingData, testData) = data.randomSplit([0.7, 0.3])
+
+    # Train a RandomForest model.
+    rf = RandomForestRegressor(featuresCol="indexedFeatures")
+
+    # Chain indexer and forest in a Pipeline
+    pipeline = Pipeline(stages=[featureIndexer, rf])
+
+    # Train model.  This also runs the indexer.
+    model = pipeline.fit(trainingData)
+
+    # Make predictions.
+    predictions = model.transform(testData)
+
+    # Select example rows to display.
+    predictions.select("prediction", "label", "features").show(5)
+
+    # Select (prediction, true label) and compute test error
+    evaluator = RegressionEvaluator(
+        labelCol="label", predictionCol="prediction", metricName="rmse")
+    rmse = evaluator.evaluate(predictions)
+    print("Root Mean Squared Error (RMSE) on test data = %g" % rmse)
+
+    rfModel = model.stages[1]
+    print(rfModel)  # summary only
+
+    spark.stop()

examples/random-forest-pyspark/run-gpu-standalone-regressor.sh

@@ -0,0 +1,38 @@
+#!/usr/bin/env bash
+
+source ../../conf/env.sh
+
+# CSV data is the same as in Spark example "ml/pca_example.py"
+# The data file should be copied to $HDFS_ROOT before running examples
+DATA_FILE=$HDFS_ROOT/data/sample_libsvm_data.txt
+
+DEVICE=GPU
+RESOURCE_FILE=$PWD/IntelGpuResourceFile.json
+WORKER_GPU_AMOUNT=4
+EXECUTOR_GPU_AMOUNT=1
+TASK_GPU_AMOUNT=1
+APP_PY=random_forest_regressor_example.py
+
+
+time $SPARK_HOME/bin/spark-submit --master $SPARK_MASTER \
+    --num-executors $SPARK_NUM_EXECUTORS \
+    --executor-cores $SPARK_EXECUTOR_CORES \
+    --total-executor-cores $SPARK_TOTAL_CORES \
+    --driver-memory $SPARK_DRIVER_MEMORY \
+    --executor-memory $SPARK_EXECUTOR_MEMORY \
+    --conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" \
+    --conf "spark.default.parallelism=$SPARK_DEFAULT_PARALLELISM" \
+    --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.worker.resourcesFile=$RESOURCE_FILE" \
+    --conf "spark.worker.resource.gpu.amount=$WORKER_GPU_AMOUNT" \
+    --conf "spark.executor.resource.gpu.amount=$EXECUTOR_GPU_AMOUNT" \
+    --conf "spark.task.resource.gpu.amount=$TASK_GPU_AMOUNT" \
+    --conf "spark.shuffle.reduceLocality.enabled=false" \
+    --conf "spark.network.timeout=1200s" \
+    --conf "spark.task.maxFailures=1" \
+    --jars $OAP_MLLIB_JAR \
+    $APP_PY DATA_FILE \
+    2>&1 | tee random_forest_regressor-$(date +%m%d_%H_%M_%S).log

mllib-dal/src/main/java/com/intel/oap/mllib/classification/RandomForestResult.java

@@ -3,4 +3,5 @@
 public class RandomForestResult {
     public long predictionNumericTable;
     public long probabilitiesNumericTable;
+    public long importancesNumericTable;
 }

mllib-dal/src/main/java/com/intel/oneapi/dal/table/HomogenTableImpl.java

@@ -15,6 +15,7 @@ public class HomogenTableImpl implements HomogenTableIface {
     private long cObject;
     private TableMetadata metadata;
     private Common.ComputeDevice device;
+
     protected HomogenTableImpl(Common.ComputeDevice computeDevice) {
         super();
         this.device = computeDevice;

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

@@ -333,6 +333,11 @@ Java_com_intel_oap_mllib_classification_RandomForestClassifierDALImpl_cRFClassif
             maxTreeDepth, seed, maxBins, bootstrap, comm, resultObj);
         return hashmapObj;
     }
+    default: {
+        std::cout << "RandomForest (native): The compute device "
+                  << "is not supported!" << std::endl;
+        exit(-1);
+    }
     }
     return nullptr;
 }