diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDesc.scala index 3127fa91232..8fd06b79e45 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDesc.scala @@ -117,8 +117,9 @@ abstract class SklearnMLOperatorDescriptor[T <: ParamClass] extends PythonOperat | self.dataset = table | | if port == 1 : - | y_train = self.dataset[$groundTruthAttribute] - | X_train = self.dataset[features] + | dataset = self.dataset.dropna(subset=features + [$groundTruthAttribute]) #remove missing values + | y_train = dataset[$groundTruthAttribute] + | X_train = dataset[features] | loop_times = ${getLoopTimes(paraList)} | | for i in range(loop_times): diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDesc.scala index 92aec692a6e..f7a2181976c 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDesc.scala @@ -39,6 +39,8 @@ abstract class SklearnClassifierOpDesc extends SklearnModelOpDesc { |class ProcessTableOperator(UDFTableOperator): | @overrides | def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]: + | table = ${if (countVectorizer) pyb"table.dropna(subset=[$text, $target])" + else "table.dropna()"} #remove missing values | Y = table[$target] | X = table.drop($target, axis=1) | X = ${if (countVectorizer) pyb"X[$text]" else "X"} diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDesc.scala index 6e894fccd90..d7c17eaac44 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDesc.scala @@ -62,6 +62,9 @@ class SklearnPredictionOpDesc extends PythonOperatorDescriptor { | input_features = tuple_ | if $groundTruthAttribute != "": | input_features = input_features.get_partial_tuple([col for col in tuple_.get_field_names() if col != $groundTruthAttribute]) + | if Table.from_tuple_likes([tuple_]).isna().any(axis=None): + | tuple_[$resultAttribute] = None #keep the row, leave the result empty + | elif $groundTruthAttribute != "": | tuple_[$resultAttribute] = type(tuple_[$groundTruthAttribute])(self.model.predict(Table.from_tuple_likes([input_features]))[0]) | else: | tuple_[$resultAttribute] = str(self.model.predict(Table.from_tuple_likes([input_features]))[0]) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDesc.scala index e262bb29531..f4204d48bdb 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDesc.scala @@ -66,7 +66,7 @@ class SklearnTestingOpDesc extends PythonOperatorDescriptor { | self.data.append(tuple_) | else: | model = tuple_[$model] - | table = Table(self.data) + | table = Table(self.data).dropna() #remove missing values | Y = table[$target] | X = table.drop($target, axis=1) | predictions = model.predict(X.squeeze()) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingOpDesc.scala index 3f354cc37f5..2bba59e2537 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingOpDesc.scala @@ -39,6 +39,8 @@ class SklearnTrainingOpDesc extends SklearnModelOpDesc { |class ProcessTableOperator(UDFTableOperator): | @overrides | def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]: + | table = ${if (countVectorizer) pyb"table.dropna(subset=[$text, $target])" + else "table.dropna()"} #remove missing values | Y = table[$target] | X = table.drop($target, axis=1) | X = ${if (countVectorizer) pyb"X[$text]" else "X"} diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDescSpec.scala index ba620af298a..f151fc3b3ff 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDescSpec.scala @@ -103,6 +103,13 @@ class SklearnAdvancedBaseDescSpec extends AnyFlatSpec with Matchers { code should include("yield df") } + // This family reads a named list of features rather than every column, so the + // drop names those columns: a blank anywhere else must not cost the row. + it should "drop rows missing a selected feature or the ground truth" in { + val d = newOp(List(hyperParam("n_neighbors", "int", fromWorkflow = false, value = "5"))) + d.generatePythonCode() should include("self.dataset.dropna(subset=features + [") + } + it should "loop once when no parameter is sourced from the workflow" in { val d = newOp(List(hyperParam("n_neighbors", "int", fromWorkflow = false, value = "5"))) val code = d.generatePythonCode() diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnBernoulliNaiveBayesOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnBernoulliNaiveBayesOpDescSpec.scala index b34d9b2e1b9..c86117419a8 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnBernoulliNaiveBayesOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnBernoulliNaiveBayesOpDescSpec.scala @@ -65,6 +65,14 @@ class SklearnBernoulliNaiveBayesOpDescSpec extends AnyFlatSpec with Matchers { code should include("Bernoulli Naive Bayes") } + // The same table statement serves both ports, so training and scoring skip a + // row with a missing value alike. + it should "drop rows with missing values before fitting and before scoring" in { + val d = new SklearnBernoulliNaiveBayesOpDesc + d.target = "y" + d.generatePythonCode() should include("table.dropna()") + } + "SklearnBernoulliNaiveBayesOpDesc" should "round-trip its config fields through the polymorphic base" in { val d = new SklearnBernoulliNaiveBayesOpDesc diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDescSpec.scala index 2b5a76284a7..c5a12eaccee 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDescSpec.scala @@ -92,6 +92,17 @@ class SklearnPredictionOpDescSpec extends AnyFlatSpec with Matchers { code should include("yield tuple_") } + // This operator adds a column to the user's rows, so a row it cannot predict + // on keeps its place with an empty result rather than disappearing. + it should "keep a row with a missing value and leave its result empty" in { + val d = new SklearnPredictionOpDesc + d.model = "model" + d.resultAttribute = "prediction" + val code = d.generatePythonCode() + code should include("isna().any(axis=None)") + code should include("] = None") + } + "SklearnPredictionOpDesc" should "round-trip its config fields through the polymorphic base" in { val d = new SklearnPredictionOpDesc diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDescSpec.scala index 8c932005003..68ec31a8d71 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDescSpec.scala @@ -80,6 +80,15 @@ class SklearnTestingOpDescSpec extends AnyFlatSpec with Matchers { code should include(".predict(") } + // The scores are computed over the rows the model can be applied to, the way + // COUNT and MIN are computed over the rows that have a value. + it should "drop rows with missing values before scoring" in { + val d = new SklearnTestingOpDesc + d.model = "model" + d.target = "y" + d.generatePythonCode() should include("Table(self.data).dropna()") + } + "SklearnTestingOpDesc" should "round-trip its config fields through the polymorphic base" in { val d = new SklearnTestingOpDesc diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingBernoulliNaiveBayesOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingBernoulliNaiveBayesOpDescSpec.scala index 7bfe9c0905b..8ffb8691df4 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingBernoulliNaiveBayesOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingBernoulliNaiveBayesOpDescSpec.scala @@ -64,6 +64,24 @@ class SklearnTrainingBernoulliNaiveBayesOpDescSpec extends AnyFlatSpec with Matc code should include("Training: Bernoulli Naive Bayes") } + // Every column but the target is a feature here, so a row missing any value is + // one the estimator cannot be fitted on. + it should "drop rows with missing values before fitting" in { + val d = new SklearnTrainingBernoulliNaiveBayesOpDesc + d.target = "y" + d.generatePythonCode() should include("table.dropna()") + } + + // With Count Vectorizer on, only the text and target columns are read, so a + // blank in any other column must not cost the row. + it should "drop on the text and target columns only when vectorizing text" in { + val d = new SklearnTrainingBernoulliNaiveBayesOpDesc + d.target = "y" + d.countVectorizer = true + d.text = "note" + d.generatePythonCode() should include("table.dropna(subset=[") + } + "SklearnTrainingBernoulliNaiveBayesOpDesc" should "round-trip its config fields through the polymorphic base" in { val d = new SklearnTrainingBernoulliNaiveBayesOpDesc d.target = "label"