Spark 3 (#970)

* feat: Update to Spark 3.0 Co-authored-by: Mark Hamilton <marhamil@microsoft.com> Co-authored-by: Daniel Ciborowski <dciborow@microsoft.com> Co-authored-by: Ilya Matiach <ilmat@microsoft.com> Co-authored-by: Markus Cozowicz <marcozo@microsoft.com> Co-authored-by: Jack Gerrits <jagerrit@microsoft.com>
microsoft · Jan 28, 2021 · 6218a5b · 6218a5b
1 parent 5a5147a
commit 6218a5b
Show file tree

Hide file tree

Showing 106 changed files with 912 additions and 1,027 deletions.
diff --git a/build.sbt b/build.sbt
@@ -11,12 +11,15 @@ import scala.sys.process.Process
 val condaEnvName = "mmlspark"
 name := "mmlspark"
 organization := "com.microsoft.ml.spark"
-scalaVersion := "2.11.12"
+scalaVersion := "2.12.10"
+val sparkVersion = "3.0.1"
 
-val sparkVersion = "2.4.5"
+//val scalaMajorVersion  = settingKey[String]("scalaMajorVersion")
+//scalaMajorVersion  := {scalaVersion.value.split(".".toCharArray).dropRight(0).mkString(".")}
+val scalaMajorVersion = 2.12
 
 val excludes = Seq(
-  ExclusionRule("org.apache.spark", "spark-tags_2.11"),
+  ExclusionRule("org.apache.spark", s"spark-tags_$scalaMajorVersion"),
   ExclusionRule("org.scalatic"),
   ExclusionRule("org.scalatest")
 )
@@ -35,9 +38,9 @@ libraryDependencies ++= Seq(
   "org.apache.httpcomponents" % "httpclient" % "4.5.6" excludeAll (excludes: _*),
   "org.apache.httpcomponents" % "httpmime" % "4.5.6" excludeAll (excludes: _*),
   "com.microsoft.ml.lightgbm" % "lightgbmlib" % "2.3.180" excludeAll (excludes: _*),
-  "com.github.vowpalwabbit" % "vw-jni" % "8.8.1" excludeAll (excludes: _*),
-  "com.linkedin.isolation-forest" %% "isolation-forest_2.4.3" % "0.3.2" excludeAll (excludes: _*),
-  "org.apache.spark" %% "spark-avro" % sparkVersion % "provided",
+  "com.github.vowpalwabbit" % "vw-jni" % "8.9.1" excludeAll (excludes: _*),
+  "com.linkedin.isolation-forest" %% "isolation-forest_3.0.0" % "1.0.1" excludeAll (excludes: _*),
+  "org.apache.spark" %% "spark-avro" % sparkVersion % "provided"
 )
 
 def txt(e: Elem, label: String): String = "\"" + e.child.filter(_.label == label).flatMap(_.text).mkString + "\""
@@ -111,9 +114,11 @@ def activateCondaEnv: Seq[String] = {
   }
 }
 
+
+
 val packagePythonTask = TaskKey[Unit]("packagePython", "Package python sdk")
-val genDir = join("target", "scala-2.11", "generated")
-val unidocDir = join("target", "scala-2.11", "unidoc")
+val genDir = join("target", s"scala-${scalaMajorVersion}", "generated")
+val unidocDir = join("target", s"scala-${scalaMajorVersion}", "unidoc")
 val pythonSrcDir = join(genDir.toString, "src", "python")
 val unifiedDocDir = join(genDir.toString, "doc")
 val pythonDocDir = join(unifiedDocDir.toString, "pyspark")
@@ -198,7 +203,7 @@ val publishR = TaskKey[Unit]("publishR", "publish R package to blob")
 publishR := {
   val s = streams.value
   (runMain in Test).toTask(" com.microsoft.ml.spark.codegen.CodeGen").value
-  val rPackage = join("target", "scala-2.11", "generated", "package", "R")
+  val rPackage = join("target", s"scala-${scalaMajorVersion}", "generated", "package", "R")
     .listFiles().head
   singleUploadToBlob(rPackage.toString, rPackage.getName, "rrr", s.log)
 }
@@ -207,7 +212,7 @@ packagePythonTask := {
   val s = streams.value
   (runMain in Test).toTask(" com.microsoft.ml.spark.codegen.CodeGen").value
   createCondaEnvTask.value
-  val destPyDir = join("target", "scala-2.11", "classes", "mmlspark")
+  val destPyDir = join("target", s"scala-${scalaMajorVersion}", "classes", "mmlspark")
   if (destPyDir.exists()) FileUtils.forceDelete(destPyDir)
   FileUtils.copyDirectory(join(pythonSrcDir.getAbsolutePath, "mmlspark"), destPyDir)
 
@@ -243,7 +248,7 @@ testPythonTask := {
       "--cov-report=xml",
       "mmlsparktest"
     ),
-    new File("target/scala-2.11/generated/test/python/"),
+    new File(s"target/scala-${scalaMajorVersion}/generated/test/python/")
   ) ! s.log
 }
 
@@ -252,7 +257,7 @@ val datasetName = "datasets-2020-08-27.tgz"
 val datasetUrl = new URL(s"https://mmlspark.blob.core.windows.net/installers/$datasetName")
 val datasetDir = settingKey[File]("The directory that holds the dataset")
 datasetDir := {
-  join(target.value.toString, "scala-2.11", "datasets", datasetName.split(".".toCharArray.head).head)
+  join(target.value.toString, s"scala-${scalaMajorVersion}", "datasets", datasetName.split(".".toCharArray.head).head)
 }
 
 getDatasetsTask := {
@@ -270,19 +275,19 @@ genBuildInfo := {
 
   val buildInfo =
     s"""
-       |MMLSpark Build and Release Information
-       |---------------
-       |
-       |### Maven Coordinates
-       | `${organization.value}:${name.value}_2.11:${version.value}`
-       |
-       |### Maven Resolver
-       | `https://mmlspark.azureedge.net/maven`
-       |
-       |### Documentation Pages:
-       |[Scala Documentation](https://mmlspark.blob.core.windows.net/docs/${version.value}/scala/index.html)
-       |[Python Documentation](https://mmlspark.blob.core.windows.net/docs/${version.value}/pyspark/index.html)
-       |
+      |MMLSpark Build and Release Information
+      |---------------
+      |
+      |### Maven Coordinates
+      | `${organization.value}:${name.value}_${scalaMajorVersion}:${version.value}`
+      |
+      |### Maven Resolver
+      | `https://mmlspark.azureedge.net/maven`
+      |
+      |### Documentation Pages:
+      |[Scala Documentation](https://mmlspark.blob.core.windows.net/docs/${version.value}/scala/index.html)
+      |[Python Documentation](https://mmlspark.blob.core.windows.net/docs/${version.value}/pyspark/index.html)
+      |
     """.stripMargin
 
   val infoFile = join("target", "Build.md")
@@ -364,7 +369,7 @@ val settings = Seq(
     case x => MergeStrategy.first
   },
   assemblyOption in assembly := (assemblyOption in assembly).value.copy(includeScala = false),
-  buildInfoPackage := "com.microsoft.ml.spark.build") //++
+  buildInfoPackage := "com.microsoft.ml.spark.build")
 
 lazy val mmlspark = (project in file("."))
   .enablePlugins(BuildInfoPlugin)
@@ -414,3 +419,5 @@ pgpPublicRing := {
 dynverSonatypeSnapshots in ThisBuild := true
 dynverSeparator in ThisBuild := "-"
 publishTo := sonatypePublishToBundle.value
+
+// Cache Break 1
diff --git a/environment.yaml b/environment.yaml
@@ -1,7 +1,9 @@
 name: mmlspark
+channels:
+  - conda-forge
 dependencies:
   - python=3.6
-  - pyspark=2.4.3
+  - pyspark=3.0.1
   - requests
   - pip:
     - wheel

diff --git a/notebooks/samples/DeepLearning - BiLSTM Medical Entity Extraction.ipynb b/notebooks/samples/DeepLearning - BiLSTM Medical Entity Extraction.ipynb
@@ -174,10 +174,16 @@
   {
    "cell_type": "code",
    "execution_count": null,
-   "metadata": {},
    "outputs": [],
    "source": [
-    "tokenizeUDF = udf(word_tokenize, ArrayType(StringType()))\n",
+    "def safe_tokenize(sent):\n",
+    "    try:\n",
+    "        return word_tokenize(sent)\n",
+    "    except LookupError:\n",
+    "        prepNLTK(None)\n",
+    "        return word_tokenize(sent)\n",
+    "\n",
+    "tokenizeUDF = udf(safe_tokenize, ArrayType(StringType()))\n",
     "df = df.withColumn(\"tokens\",tokenizeUDF(\"sentence\"))\n",
     "\n",
     "countUDF = udf(len, IntegerType())\n",
@@ -203,9 +209,15 @@
     "featurizeUDF = udf(featurize,  ArrayType(FloatType()))\n",
     "\n",
     "df = df.withColumn(\"features\", featurizeUDF(\"tokens\")).cache()\n",
-    "safe_show(df, 3) # Can be flaky on build server\n",
+    "safe_show(df, 5) # Can be flaky on build server\n",
     "    \n"
-   ]
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%%\n"
+    }
+   }
   },
   {
    "cell_type": "markdown",
@@ -338,15 +350,6 @@
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
    "version": "3.6.3"
-  },
-  "pycharm": {
-   "stem_cell": {
-    "cell_type": "raw",
-    "source": [],
-    "metadata": {
-     "collapsed": false
-    }
-   }
   }
  },
  "nbformat": 4,

diff --git a/notebooks/samples/LightGBM - Quantile Regression for Drug Discovery.ipynb b/notebooks/samples/LightGBM - Quantile Regression for Drug Discovery.ipynb
@@ -94,8 +94,8 @@
    "outputs": [],
    "source": [
     "from mmlspark.lightgbm import LightGBMRegressionModel\n",
-    "model.saveNativeModel(\"mymodel\")\n",
-    "model = LightGBMRegressionModel.loadNativeModelFromFile(\"mymodel\")"
+    "model.saveNativeModel(\"/mymodel\")\n",
+    "model = LightGBMRegressionModel.loadNativeModelFromFile(\"/mymodel\")"
    ]
   },
   {
@@ -175,4 +175,4 @@
  },
  "nbformat": 4,
  "nbformat_minor": 2
-}
+}
diff --git a/notebooks/samples/Regression - Flight Delays with DataCleaning.ipynb b/notebooks/samples/Regression - Flight Delays with DataCleaning.ipynb
@@ -153,7 +153,7 @@
     "testCat  = DataConversion(cols=[\"Carrier\",\"DepTimeBlk\",\"ArrTimeBlk\"],\n",
     "                          convertTo=\"toCategorical\") \\\n",
     "               .transform(test)\n",
-    "lr = LinearRegression().setSolver(\"l-bfgs\").setRegParam(0.1) \\\n",
+    "lr = LinearRegression().setRegParam(0.1) \\\n",
     "                       .setElasticNetParam(0.3)\n",
     "model = TrainRegressor(model=lr, labelCol=\"ArrDelay\").fit(trainCat)"
    ]
@@ -232,15 +232,6 @@
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
    "version": "3.6.3"
-  },
-  "pycharm": {
-   "stem_cell": {
-    "cell_type": "raw",
-    "source": [],
-    "metadata": {
-     "collapsed": false
-    }
-   }
   }
  },
  "nbformat": 4,

diff --git a/pipeline.yaml b/pipeline.yaml
@@ -250,7 +250,7 @@ jobs:
         chmod +x .codecov
         echo "Starting Codecov Upload"
         ./.codecov -t $(codecov-token) -f coverage.xml
-        ./.codecov -t $(codecov-token) -f target/scala-2.11/coverage-report/cobertura.xml
+        ./.codecov -t $(codecov-token) -f target/scala-2.12/coverage-report/cobertura.xml
       displayName: Upload Coverage Report To Codecov.io
       condition: succeededOrFailed()
 

diff --git a/src/main/python/mmlspark/cyber/feature/indexers.py b/src/main/python/mmlspark/cyber/feature/indexers.py
@@ -2,14 +2,14 @@
 
 from typing import List
 
-from mmlspark.cyber.utils.spark_utils import DataFrameUtils, ExplainBuilder
+from mmlspark.cyber.utils.spark_utils import DataFrameUtils, ExplainBuilder, HasSetInputCol, HasSetOutputCol
 
 from pyspark.ml import Estimator, Transformer
 from pyspark.ml.param.shared import HasInputCol, HasOutputCol, Param, Params
 from pyspark.sql import DataFrame, functions as f
 
 
-class IdIndexerModel(Transformer, HasInputCol, HasOutputCol):
+class IdIndexerModel(Transformer, HasSetInputCol, HasSetOutputCol):
     partitionKey = Param(
         Params._dummy(),
         "partitionKey",
@@ -43,7 +43,7 @@ def _transform(self, df):
         )
 
 
-class IdIndexer(Estimator, HasInputCol, HasOutputCol):
+class IdIndexer(Estimator, HasSetInputCol, HasSetOutputCol):
     partitionKey = Param(
         Params._dummy(),
         "partitionKey",

diff --git a/src/main/python/mmlspark/cyber/feature/scalers.py b/src/main/python/mmlspark/cyber/feature/scalers.py
@@ -3,7 +3,7 @@
 from abc import ABC, abstractmethod
 from typing import Callable, Dict, List, Optional, Union
 
-from mmlspark.cyber.utils.spark_utils import ExplainBuilder
+from mmlspark.cyber.utils.spark_utils import ExplainBuilder, HasSetInputCol, HasSetOutputCol
 
 from pyspark.ml import Estimator, Transformer
 from pyspark.ml.param.shared import HasInputCol, HasOutputCol, Param, Params
@@ -15,7 +15,7 @@ def _pyudf(func, use_pandas):
     return pandas_udf(func, t.DoubleType(), PandasUDFType.SCALAR) if use_pandas else udf(func, t.DoubleType())
 
 
-class PerPartitionScalarScalerModel(ABC, Transformer, HasInputCol, HasOutputCol):
+class PerPartitionScalarScalerModel(ABC, Transformer, HasSetInputCol, HasSetOutputCol):
     partitionKey = Param(
         Params._dummy(),
         "partitionKey",
@@ -83,7 +83,7 @@ def _transform(self, df: DataFrame) -> DataFrame:
         return with_stats_df.withColumn(output_col, stats_method(f.col(input_col)))
 
 
-class PerPartitionScalarScalerEstimator(ABC, Estimator, HasInputCol, HasOutputCol):
+class PerPartitionScalarScalerEstimator(ABC, Estimator, HasSetInputCol, HasSetOutputCol):
     partitionKey = Param(
         Params._dummy(),
         "partitionKey",

diff --git a/src/main/python/mmlspark/cyber/utils/spark_utils.py b/src/main/python/mmlspark/cyber/utils/spark_utils.py
@@ -175,3 +175,17 @@ def build(explainable: Any, **kwargs):
 
         # noinspection PyProtectedMember
         explainable._set(**kwargs)
+
+class HasSetInputCol(HasInputCol):
+    def setInputCol(self, value):
+        """
+        Sets the value of :py:attr:`inputCol`.
+        """
+        return self.set(self.inputCol, value)
+
+class HasSetOutputCol(HasOutputCol):
+    def setOutputCol(self, value):
+        """
+        Sets the value of :py:attr:`outputCol`.
+        """
+        return self.set(self.outputCol, value)
diff --git a/src/main/python/mmlspark/recommendation/RankingTrainValidationSplit.py b/src/main/python/mmlspark/recommendation/RankingTrainValidationSplit.py
@@ -6,7 +6,7 @@
 from mmlspark.recommendation._RankingTrainValidationSplit import _RankingTrainValidationSplit
 from pyspark import keyword_only
 from pyspark.ml.param import Params
-from pyspark.ml.tuning import ValidatorParams
+from pyspark.ml.tuning import _ValidatorParams
 from pyspark.ml.util import *
 from pyspark import keyword_only
 from pyspark.ml.param import Params, Param, TypeConverters
@@ -20,7 +20,7 @@
 
 
 @inherit_doc
-class RankingTrainValidationSplit(Estimator, ValidatorParams):
+class RankingTrainValidationSplit(Estimator, _ValidatorParams):
     trainRatio = Param(Params._dummy(), "trainRatio", "Param for ratio between train and\
          validation data. Must be between 0 and 1.", typeConverter=TypeConverters.toFloat)
     userCol = Param(Params._dummy(), "userCol",

diff --git a/src/main/python/mmlspark/recommendation/RankingTrainValidationSplitModel.py b/src/main/python/mmlspark/recommendation/RankingTrainValidationSplitModel.py
@@ -7,7 +7,7 @@
     basestring = str
 
 from pyspark.ml.common import inherit_doc
-from pyspark.ml.tuning import ValidatorParams
+from pyspark.ml.tuning import _ValidatorParams
 from pyspark.ml.util import *
 from mmlspark.recommendation._RankingTrainValidationSplitModel import _RankingTrainValidationSplitModel
 from pyspark.ml.wrapper import JavaParams
@@ -16,7 +16,7 @@
 
 # Load information from java_stage to the instance.
 @inherit_doc
-class RankingTrainValidationSplitModel(_RankingTrainValidationSplitModel, ValidatorParams):
+class RankingTrainValidationSplitModel(_RankingTrainValidationSplitModel, _ValidatorParams):
 
     def __init__(self, bestModel=None, validationMetrics=[]):
         super(RankingTrainValidationSplitModel, self).__init__()

diff --git a/src/main/scala/com/microsoft/ml/spark/automl/HyperparamBuilder.scala b/src/main/scala/com/microsoft/ml/spark/automl/HyperparamBuilder.scala
@@ -5,7 +5,7 @@ package com.microsoft.ml.spark.automl
 
 import org.apache.spark.ml.param._
 
-import scala.collection.{JavaConversions, mutable}
+import scala.collection.{JavaConverters, mutable}
 import scala.util.Random
 
 abstract class RangeHyperParam[T](val min: T, val max: T, val seed: Long) extends Dist[T] {
@@ -77,7 +77,7 @@ object HyperParamUtils {
     * @return A RangeHyperParam matched to the given type for min and max values.
     */
   def getDiscreteHyperParam(values: java.util.ArrayList[_], seed: Long = 0): DiscreteHyperParam[_] = {
-    val valuesList = JavaConversions.asScalaBuffer(values).toList
+    val valuesList = JavaConverters.asScalaBuffer(values).toList
     new DiscreteHyperParam(valuesList, seed)
   }
 }