feat: add base classes for ML and refine code base (#1031)

Author: serena-ruan

src/csharp/Microsoft.Spark.E2ETest/IpcTests/ML/Feature/FeatureBaseTests.cs

@@ -25,7 +25,7 @@ protected FeatureBaseTests(SparkFixture fixture)
         /// <param name="paramName">The name of a parameter that can be set on this object</param>
         /// <param name="paramValue">A parameter value that can be set on this object</param>
         public void TestFeatureBase(
-            FeatureBase<T> testObject, 
+            Params testObject, 
             string paramName, 
             object paramValue)
         {
@@ -37,8 +37,8 @@ public void TestFeatureBase(
             Assert.Equal(param.Parent, testObject.Uid());
 
             Assert.NotEmpty(testObject.ExplainParam(param));
-            testObject.Set(param, paramValue);
-            Assert.IsAssignableFrom<Identifiable>(testObject.Clear(param));
+            testObject.Set<T>(param, paramValue);
+            Assert.IsAssignableFrom<Identifiable>(testObject.Clear<T>(param));
 
             Assert.IsType<string>(testObject.Uid());
         }

src/csharp/Microsoft.Spark.E2ETest/IpcTests/ML/Feature/PipelineModelTests.cs

@@ -0,0 +1,79 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.IO;
+using Microsoft.Spark.ML.Feature;
+using Microsoft.Spark.Sql;
+using Microsoft.Spark.UnitTest.TestUtils;
+using Microsoft.Spark.Sql.Types;
+using Xunit;
+
+namespace Microsoft.Spark.E2ETest.IpcTests.ML.Feature
+{
+    [Collection("Spark E2E Tests")]
+    public class PipelineModelTests : FeatureBaseTests<PipelineModel>
+    {
+        private readonly SparkSession _spark;
+
+        public PipelineModelTests(SparkFixture fixture) : base(fixture)
+        {
+            _spark = fixture.Spark;
+        }
+
+        /// <summary>
+        /// Create a <see cref="PipelineModel"/> and test the
+        /// available methods.
+        /// </summary>
+        [Fact]
+        public void TestPipelineModelTransform()
+        {
+            var expectedSplits =
+                new double[] { double.MinValue, 0.0, 10.0, 50.0, double.MaxValue };
+
+            string expectedHandle = "skip";
+            string expectedUid = "uid";
+            string expectedInputCol = "input_col";
+            string expectedOutputCol = "output_col";
+
+            var bucketizer = new Bucketizer(expectedUid);
+            bucketizer.SetInputCol(expectedInputCol)
+                .SetOutputCol(expectedOutputCol)
+                .SetHandleInvalid(expectedHandle)
+                .SetSplits(expectedSplits);
+
+            var stages = new JavaTransformer[] {
+                bucketizer
+            };
+
+            PipelineModel pipelineModel = new PipelineModel("randomUID", stages);
+
+            DataFrame input = _spark.Sql("SELECT ID as input_col from range(100)");
+
+            DataFrame output = pipelineModel.Transform(input);
+            Assert.Contains(output.Schema().Fields, (f => f.Name == expectedOutputCol));
+
+            Assert.Equal(expectedInputCol, bucketizer.GetInputCol());
+            Assert.Equal(expectedOutputCol, bucketizer.GetOutputCol());
+            Assert.Equal(expectedSplits, bucketizer.GetSplits());
+
+            Assert.IsType<StructType>(pipelineModel.TransformSchema(input.Schema()));
+            Assert.IsType<DataFrame>(output);
+            
+            using (var tempDirectory = new TemporaryDirectory())
+            {
+                string savePath = Path.Join(tempDirectory.Path, "pipelineModel");
+                pipelineModel.Save(savePath);
+
+                PipelineModel loadedPipelineModel = PipelineModel.Load(savePath);
+                Assert.Equal(pipelineModel.Uid(), loadedPipelineModel.Uid());
+
+                string writePath = Path.Join(tempDirectory.Path, "pipelineModelWithWrite");
+                pipelineModel.Write().Save(writePath);
+
+                PipelineModel loadedPipelineModelWithRead = pipelineModel.Read().Load(writePath);
+                Assert.Equal(pipelineModel.Uid(), loadedPipelineModelWithRead.Uid());
+            }
+        }
+    }
+}

src/csharp/Microsoft.Spark.E2ETest/IpcTests/ML/Feature/PipelineTests.cs

@@ -0,0 +1,111 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.IO;
+using Microsoft.Spark.ML.Feature;
+using Microsoft.Spark.Sql;
+using Microsoft.Spark.UnitTest.TestUtils;
+using Microsoft.Spark.Sql.Types;
+using Xunit;
+
+namespace Microsoft.Spark.E2ETest.IpcTests.ML.Feature
+{
+    [Collection("Spark E2E Tests")]
+    public class PipelineTests : FeatureBaseTests<Pipeline>
+    {
+        private readonly SparkSession _spark;
+
+        public PipelineTests(SparkFixture fixture) : base(fixture)
+        {
+            _spark = fixture.Spark;
+        }
+
+        /// <summary>
+        /// Create a <see cref="Pipeline"/> and test the
+        /// available methods. Test the FeatureBase methods 
+        /// using <see cref="TestFeatureBase"/>.
+        /// </summary>
+        [Fact]
+        public void TestPipeline()
+        {
+            var stages = new JavaPipelineStage[] {
+                new Bucketizer(),
+                new CountVectorizer()
+            };
+
+            Pipeline pipeline = new Pipeline()
+                .SetStages(stages);
+            JavaPipelineStage[] returnStages = pipeline.GetStages();
+            
+            Assert.Equal(stages[0].Uid(), returnStages[0].Uid());
+            Assert.Equal(stages[0].ToString(), returnStages[0].ToString());
+            Assert.Equal(stages[1].Uid(), returnStages[1].Uid());
+            Assert.Equal(stages[1].ToString(), returnStages[1].ToString());
+            
+            using (var tempDirectory = new TemporaryDirectory())
+            {
+                string savePath = Path.Join(tempDirectory.Path, "pipeline");
+                pipeline.Save(savePath);
+
+                Pipeline loadedPipeline = Pipeline.Load(savePath);
+                Assert.Equal(pipeline.Uid(), loadedPipeline.Uid());
+            }
+            
+            TestFeatureBase(pipeline, "stages", stages);
+        }
+
+        /// <summary>
+        /// Create a <see cref="Pipeline"/> and test the
+        /// fit and read/write methods.
+        /// </summary>
+        [Fact]
+        public void TestPipelineFit()
+        {
+            DataFrame input = _spark.Sql("SELECT array('hello', 'I', 'AM', 'a', 'string', 'TO', " +
+                "'TOKENIZE') as input from range(100)");
+
+            const string inputColumn = "input";
+            const string outputColumn = "output";
+            const double minDf = 1;
+            const double minTf = 10;
+            const int vocabSize = 10000;
+            
+            CountVectorizer countVectorizer = new CountVectorizer()
+                .SetInputCol(inputColumn)
+                .SetOutputCol(outputColumn)
+                .SetMinDF(minDf)
+                .SetMinTF(minTf)
+                .SetVocabSize(vocabSize);
+
+            var stages = new JavaPipelineStage[] {
+                countVectorizer
+            };
+
+            Pipeline pipeline = new Pipeline().SetStages(stages);
+            PipelineModel pipelineModel = pipeline.Fit(input);
+            
+            DataFrame output = pipelineModel.Transform(input);
+
+            Assert.IsType<StructType>(pipelineModel.TransformSchema(input.Schema()));
+            Assert.IsType<DataFrame>(output);
+            
+            using (var tempDirectory = new TemporaryDirectory())
+            {
+                string savePath = Path.Join(tempDirectory.Path, "pipeline");
+                pipeline.Save(savePath);
+
+                Pipeline loadedPipeline = Pipeline.Load(savePath);
+                Assert.Equal(pipeline.Uid(), loadedPipeline.Uid());
+
+                string writePath = Path.Join(tempDirectory.Path, "pipelineWithWrite");
+                pipeline.Write().Save(writePath);
+
+                Pipeline loadedPipelineWithRead = pipeline.Read().Load(writePath);
+                Assert.Equal(pipeline.Uid(), loadedPipelineWithRead.Uid());
+            }
+            
+            TestFeatureBase(pipeline, "stages", stages);
+        }
+    }
+}

src/csharp/Microsoft.Spark/Interop/Internal/Dotnet/ArrayExtensions.cs

@@ -0,0 +1,33 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using Microsoft.Spark.Interop;
+using Microsoft.Spark.Interop.Internal.Java.Util;
+
+namespace System
+{
+    /// <summary>
+    /// ArrayExtensions host custom extension methods for the
+    /// dotnet base class array T[].
+    /// </summary>
+    public static class ArrayExtensions
+    {
+        /// <summary>
+        /// A custom extension method that helps transform from dotnet
+        /// array of type T to java.util.ArrayList.
+        /// </summary>
+        /// <param name="array">an array instance</param>
+        /// <typeparam name="T">elements type of param array</typeparam>
+        /// <returns><see cref="ArrayList"/></returns>
+        internal static ArrayList ToJavaArrayList<T>(this T[] array)
+        {
+            var arrayList = new ArrayList(SparkEnvironment.JvmBridge);
+            foreach (T item in array)
+            {
+                arrayList.Add(item);
+            }
+            return arrayList;
+        }
+    }
+}

src/csharp/Microsoft.Spark/Interop/Internal/Dotnet/DictionaryExtensions.cs

@@ -0,0 +1,44 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using Microsoft.Spark.Interop;
+using Microsoft.Spark.Interop.Internal.Java.Util;
+
+namespace System.Collections.Generic
+{
+    public static class Dictionary
+    {
+        /// <summary>
+        /// A custom extension method that helps transform from dotnet
+        /// Dictionary&lt;string, string&gt; to java.util.HashMap.
+        /// </summary>
+        /// <param name="dictionary">a Dictionary instance</param>
+        /// <returns><see cref="HashMap"/></returns>
+        internal static HashMap ToJavaHashMap(this Dictionary<string, string> dictionary)
+        {
+            var hashMap = new HashMap(SparkEnvironment.JvmBridge);
+            foreach (KeyValuePair<string, string> item in dictionary)
+            {
+                hashMap.Put(item.Key, item.Value);
+            }
+            return hashMap;
+        }
+
+        /// <summary>
+        /// A custom extension method that helps transform from dotnet
+        /// Dictionary&lt;string, object&gt; to java.util.HashMap.
+        /// </summary>
+        /// <param name="dictionary">a Dictionary instance</param>
+        /// <returns><see cref="HashMap"/></returns>
+        internal static HashMap ToJavaHashMap(this Dictionary<string, object> dictionary)
+        {
+            var hashMap = new HashMap(SparkEnvironment.JvmBridge);
+            foreach (KeyValuePair<string, object> item in dictionary)
+            {
+                hashMap.Put(item.Key, item.Value);
+            }
+            return hashMap;
+        }
+    }
+}

src/csharp/Microsoft.Spark/Interop/Internal/Java/Util/HashMap.cs

@@ -0,0 +1,60 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using Microsoft.Spark.Interop.Ipc;
+
+namespace Microsoft.Spark.Interop.Internal.Java.Util
+{
+    /// <summary>
+    /// HashMap class represents a <c>java.util.HashMap</c> object.
+    /// </summary>
+    internal sealed class HashMap : IJvmObjectReferenceProvider
+    {
+        /// <summary>
+        /// Create a <c>java.util.HashMap</c> JVM object
+        /// </summary>
+        /// <param name="jvm">JVM bridge to use</param>
+        internal HashMap(IJvmBridge jvm) =>
+            Reference = jvm.CallConstructor("java.util.HashMap");
+
+        public JvmObjectReference Reference { get; private set; }
+
+        /// <summary>
+        /// Associates the specified value with the specified key in this map. 
+        /// If the map previously contained a mapping for the key, the old value is replaced.
+        /// </summary>
+        /// <param name="key">key with which the specified value is to be associated</param>
+        /// <param name="value">value to be associated with the specified key</param>
+        internal void Put(object key, object value) =>
+            Reference.Invoke("put", key, value);
+        
+        /// <summary>
+        /// Returns the value to which the specified key is mapped, 
+        /// or null if this map contains no mapping for the key.
+        /// </summary>
+        /// <param name="key">value whose presence in this map is to be tested</param>
+        /// <return>value associated with the specified key</return>
+        internal object Get(object key) =>
+            Reference.Invoke("get", key);
+
+        /// <summary>
+        /// Returns true if this map maps one or more keys to the specified value.
+        /// </summary>
+        /// <param name="value">The HashMap key</param>
+        /// <return>true if this map maps one or more keys to the specified value</return>
+        internal bool ContainsValue(object value) =>
+            (bool)Reference.Invoke("containsValue", value);
+
+        /// <summary>
+        /// Returns an array of the keys contained in this map.
+        /// </summary>
+        /// <return>An array of object hosting the keys contained in the map</return>
+        internal object[] Keys()
+        {
+            var jvmObject = (JvmObjectReference)Reference.Invoke("keySet");
+            var result = (object[])jvmObject.Invoke("toArray");
+            return result;
+        }
+    }
+}