From a717df187c407f36b63a8abb00c5d3d4855156cd Mon Sep 17 00:00:00 2001 From: kary zheng Date: Wed, 2 Sep 2026 16:32:56 -0700 Subject: [PATCH 1/2] feat(operator): export the estimators and the Hugging Face models The scikit-learn estimators are fitted on one port and score on the other, so the script holds both frames and narrows each by the same rule: a fit and a score taken on different columns would compare two different models. The four Hugging Face models declare the column types they take, and the iris regression keeps the row when a petal measurement is empty rather than ending the run on it. Sklearn Prediction and Sklearn Testing are reported as unverifiable rather than exported blind: each consumes a fitted model on an input port, and a fixture written from the JVM cannot carry a live Python object. Co-Authored-By: Claude Opus 5 (1M context) --- ...gingFaceIrisLogisticRegressionOpDesc.scala | 82 ++++++++++++++++++- .../HuggingFaceSentimentAnalysisOpDesc.scala | 64 ++++++++++++++- .../HuggingFaceSpamSMSDetectionOpDesc.scala | 51 +++++++++++- .../HuggingFaceTextSummarizationOpDesc.scala | 55 ++++++++++++- .../sklearn/SklearnClassifierOpDesc.scala | 46 ++++++++++- .../SklearnLinearRegressionOpDesc.scala | 48 ++++++++++- .../operator/sklearn/SklearnModelOpDesc.scala | 33 +++++++- .../sklearn/SklearnPredictionOpDesc.scala | 40 ++++++++- .../testing/SklearnTestingOpDesc.scala | 36 +++++++- .../training/SklearnTrainingOpDesc.scala | 30 ++++++- ...FaceIrisLogisticRegressionOpDescSpec.scala | 25 ++++++ ...ggingFaceSentimentAnalysisOpDescSpec.scala | 11 +++ ...uggingFaceSpamSMSDetectionOpDescSpec.scala | 11 +++ ...ggingFaceTextSummarizationOpDescSpec.scala | 11 +++ .../SklearnLinearRegressionOpDescSpec.scala | 13 +++ .../sklearn/SklearnModelOpDescSpec.scala | 28 +++++++ .../sklearn/SklearnPredictionOpDescSpec.scala | 21 ++++- .../testing/SklearnTestingOpDescSpec.scala | 14 ++++ 18 files changed, 590 insertions(+), 29 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceIrisLogisticRegressionOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceIrisLogisticRegressionOpDesc.scala index 9a9ac563250..347d3457ef4 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceIrisLogisticRegressionOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceIrisLogisticRegressionOpDesc.scala @@ -20,24 +20,40 @@ package org.apache.texera.amber.operator.huggingFace import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} +import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInject import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.pybuilder.PyStringTypes.EncodableString import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity} -import org.apache.texera.amber.operator.PythonOperatorDescriptor -import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName +import org.apache.texera.amber.operator.{PythonOperatorDescriptor, StandaloneCodeGenerator} +import org.apache.texera.amber.operator.metadata.annotations.{AutofillAttributeName, SampleColumn} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral -class HuggingFaceIrisLogisticRegressionOpDesc extends PythonOperatorDescriptor { +// type constraint: both measurements are standardized against the training means +// and handed to the model as floats, so each column can only be numeric. +@JsonSchemaInject(json = """ +{ + "attributeTypeRules": { + "petalLengthCmAttribute": { "enum": ["integer", "long", "double"] }, + "petalWidthCmAttribute": { "enum": ["integer", "long", "double"] } + } +} +""") +class HuggingFaceIrisLogisticRegressionOpDesc + extends PythonOperatorDescriptor + with StandaloneCodeGenerator { @JsonProperty(value = "petalLengthCmAttribute", required = true) @JsonPropertyDescription("attribute in your dataset corresponding to PetalLengthCm") @AutofillAttributeName + @SampleColumn("petal_length") var petalLengthCmAttribute: EncodableString = _ @JsonProperty(value = "petalWidthCmAttribute", required = true) @JsonPropertyDescription("attribute in your dataset corresponding to PetalWidthCm") @AutofillAttributeName + @SampleColumn("petal_width") var petalWidthCmAttribute: EncodableString = _ @JsonProperty( @@ -91,6 +107,15 @@ class HuggingFaceIrisLogisticRegressionOpDesc extends PythonOperatorDescriptor { | training_features_stds = [1.72528903, 0.73788937] | length = tuple_[$petalLengthCmAttribute] | width = tuple_[$petalWidthCmAttribute] + | # An empty cell arrives as None, which numpy carries as an object the + | # standardization cannot subtract from. Keep the row and leave the + | # prediction empty rather than ending the run over a measurement the + | # model was never given. + | if length is None or width is None: + | tuple_[$predictionClassName] = None + | tuple_[$predictionProbabilityName] = None + | yield tuple_ + | return | features = np.array([[length, width]]) | features = ((features - training_features_means) / training_features_stds) | features = torch.from_numpy(features).float() @@ -103,6 +128,57 @@ class HuggingFaceIrisLogisticRegressionOpDesc extends PythonOperatorDescriptor { | yield tuple_""".encode } + override def producesDataFrame(): Boolean = true + + // Standalone mirror of generatePythonCode: rebuild+load the pretrained linear + // model once, then apply the same per-row standardize→sigmoid→threshold logic, + // adding the STRING predicted class and DOUBLE probability columns (in + // getOutputSchemas order) to produce out1df. + override def generateStandaloneCode(): String = { + val lengthLit = pyStringLiteral(petalLengthCmAttribute) + val widthLit = pyStringLiteral(petalWidthCmAttribute) + s"""import numpy as np + |import torch + |import torch.nn as nn + |from huggingface_hub import PyTorchModelHubMixin + | + |class LinearModel(nn.Module, PyTorchModelHubMixin): + | def __init__(self): + | super().__init__() + | self.fc = nn.Linear(2, 1) + | + | def forward(self, x): + | return self.fc(x) + | + |model = LinearModel.from_pretrained("sadhaklal/logistic-regression-iris") + |model.eval() + | + |training_features_means = [3.72666667, 1.17619048] + |training_features_stds = [1.72528903, 0.73788937] + |out1df = in1df.copy() + |_classes = [] + |_probs = [] + |for _length, _width in zip(out1df[$lengthLit], out1df[$widthLit]): + | # The operator's guard, except that an empty cell reaches a frame read + | # from JSON as a NaN rather than as the None a Tuple hands over, so this + | # side asks pandas, which answers for both. + | if pd.isna(_length) or pd.isna(_width): + | _classes.append(None) + | _probs.append(None) + | continue + | features = np.array([[_length, _width]]) + | features = ((features - training_features_means) / training_features_stds) + | features = torch.from_numpy(features).float() + | with torch.no_grad(): + | logits = model(features) + | proba = torch.sigmoid(logits.squeeze()) + | preds = (proba > 0.5).long() + | _probs.append(float(proba)) + | _classes.append("Iris-setosa" if preds == 1 else "Not Iris-setosa") + |out1df[${pyStringLiteral(predictionClassName)}] = _classes + |out1df[${pyStringLiteral(predictionProbabilityName)}] = _probs""".stripMargin + } + override def operatorInfo: OperatorInfo = OperatorInfo( "Hugging Face Iris Logistic Regression", diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSentimentAnalysisOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSentimentAnalysisOpDesc.scala index 0a8fb8a8948..4e237adab44 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSentimentAnalysisOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSentimentAnalysisOpDesc.scala @@ -20,17 +20,33 @@ package org.apache.texera.amber.operator.huggingFace import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} +import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInject import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity} -import org.apache.texera.amber.operator.PythonOperatorDescriptor -import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName +import org.apache.texera.amber.operator.{PythonOperatorDescriptor, StandaloneCodeGenerator} +import org.apache.texera.amber.operator.metadata.annotations.{AutofillAttributeName, SampleColumn} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} import org.apache.texera.amber.pybuilder.PyStringTypes.EncodableString -import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext -class HuggingFaceSentimentAnalysisOpDesc extends PythonOperatorDescriptor { +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.{ + PythonTemplateBuilderStringContext, + pyStringLiteral +} +// type constraint: the tokenizer scores text and refuses anything that is not a +// string, so the column can only be a string. +@JsonSchemaInject(json = """ +{ + "attributeTypeRules": { + "attribute": { "enum": ["string"] } + } +} +""") +class HuggingFaceSentimentAnalysisOpDesc + extends PythonOperatorDescriptor + with StandaloneCodeGenerator { @JsonProperty(value = "attribute", required = true) @JsonPropertyDescription("column to perform sentiment analysis on") @AutofillAttributeName + @SampleColumn("short_text") var attribute: EncodableString = _ @JsonProperty( @@ -96,6 +112,46 @@ class HuggingFaceSentimentAnalysisOpDesc extends PythonOperatorDescriptor { | yield tuple_""".encode } + override def producesDataFrame(): Boolean = true + + // Standalone mirror of generatePythonCode: load the model once, then apply the + // same per-row softmax-over-3-labels logic to in1df, adding the three DOUBLE + // result columns (in the same order as getOutputSchemas) to produce out1df. + override def generateStandaloneCode(): String = { + val positiveLit = pyStringLiteral(resultAttributePositive) + val neutralLit = pyStringLiteral(resultAttributeNeutral) + val negativeLit = pyStringLiteral(resultAttributeNegative) + s"""from transformers import AutoModelForSequenceClassification + |from transformers import AutoTokenizer, AutoConfig + |import numpy as np + |from scipy.special import softmax + | + |model_name = "cardiffnlp/twitter-roberta-base-sentiment-latest" + |tokenizer = AutoTokenizer.from_pretrained(model_name) + |config = AutoConfig.from_pretrained(model_name) + |model = AutoModelForSequenceClassification.from_pretrained(model_name) + | + |out1df = in1df.copy() + |labels = {"positive": $positiveLit, "neutral": $neutralLit, "negative": $negativeLit} + |for _col in ($positiveLit, $neutralLit, $negativeLit): + | out1df[_col] = 0.0 + |for _idx, _text in out1df[${pyStringLiteral(attribute)}].items(): + | # An empty cell arrives as None, which the tokenizer rejects. Keep the row + | # and leave the scores empty rather than ending the run over a value the + | # model has nothing to say about. + | if _text is None or (isinstance(_text, str) and not _text.strip()): + | for _col in ($positiveLit, $neutralLit, $negativeLit): + | out1df.at[_idx, _col] = None + | continue + | encoded_input = tokenizer(_text, return_tensors='pt') + | output = model(**encoded_input) + | scores = softmax(output[0][0].detach().numpy()) + | ranking = np.argsort(scores)[::-1] + | for i in range(scores.shape[0]): + | label = labels[config.id2label[ranking[i]]] + | out1df.at[_idx, label] = np.round(float(scores[ranking[i]]), 4)""".stripMargin + } + override def operatorInfo: OperatorInfo = OperatorInfo( "Hugging Face Sentiment Analysis", diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSpamSMSDetectionOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSpamSMSDetectionOpDesc.scala index 0daa7cd4aa8..df3e5b1b277 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSpamSMSDetectionOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSpamSMSDetectionOpDesc.scala @@ -20,17 +20,33 @@ package org.apache.texera.amber.operator.huggingFace import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} +import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInject import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity} -import org.apache.texera.amber.operator.PythonOperatorDescriptor -import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName +import org.apache.texera.amber.operator.{PythonOperatorDescriptor, StandaloneCodeGenerator} +import org.apache.texera.amber.operator.metadata.annotations.{AutofillAttributeName, SampleColumn} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} import org.apache.texera.amber.pybuilder.PyStringTypes.EncodableString -import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext -class HuggingFaceSpamSMSDetectionOpDesc extends PythonOperatorDescriptor { +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.{ + PythonTemplateBuilderStringContext, + pyStringLiteral +} +// type constraint: the classification pipeline reads text and refuses anything +// that is not a string, so the column can only be a string. +@JsonSchemaInject(json = """ +{ + "attributeTypeRules": { + "attribute": { "enum": ["string"] } + } +} +""") +class HuggingFaceSpamSMSDetectionOpDesc + extends PythonOperatorDescriptor + with StandaloneCodeGenerator { @JsonProperty(value = "attribute", required = true) @JsonPropertyDescription("column to perform spam detection on") @AutofillAttributeName + @SampleColumn("short_text") var attribute: EncodableString = _ @JsonProperty( @@ -75,6 +91,33 @@ class HuggingFaceSpamSMSDetectionOpDesc extends PythonOperatorDescriptor { | yield tuple_""".encode } + override def producesDataFrame(): Boolean = true + + // Standalone mirror of generatePythonCode: build the text-classification + // pipeline once, run it per row, and add the BOOLEAN spam flag (LABEL_1) and + // the DOUBLE score columns (in getOutputSchemas order) to produce out1df. + override def generateStandaloneCode(): String = { + val attributeLit = pyStringLiteral(attribute) + val spamLit = pyStringLiteral(resultAttributeSpam) + val probabilityLit = pyStringLiteral(resultAttributeProbability) + s"""from transformers import pipeline + | + |_pipeline = pipeline("text-classification", model="mrm8488/bert-tiny-finetuned-sms-spam-detection") + |out1df = in1df.copy() + | + |def _classify(_t): + | # An empty cell arrives as None, which the pipeline rejects. Keep the row + | # and leave the results empty rather than ending the run over a value the + | # model has nothing to say about. + | if _t is None or (isinstance(_t, str) and not _t.strip()): + | return None + | return _pipeline(_t)[0] + | + |_results = [_classify(_t) for _t in out1df[$attributeLit]] + |out1df[$spamLit] = [None if _r is None else _r["label"] == "LABEL_1" for _r in _results] + |out1df[$probabilityLit] = [None if _r is None else _r["score"] for _r in _results]""".stripMargin + } + override def operatorInfo: OperatorInfo = OperatorInfo( "Hugging Face Spam Detection", diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceTextSummarizationOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceTextSummarizationOpDesc.scala index 541bcf46ac9..494cdda5ea0 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceTextSummarizationOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceTextSummarizationOpDesc.scala @@ -20,17 +20,33 @@ package org.apache.texera.amber.operator.huggingFace import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} +import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInject import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity} -import org.apache.texera.amber.operator.PythonOperatorDescriptor -import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName +import org.apache.texera.amber.operator.{PythonOperatorDescriptor, StandaloneCodeGenerator} +import org.apache.texera.amber.operator.metadata.annotations.{AutofillAttributeName, SampleColumn} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} import org.apache.texera.amber.pybuilder.PyStringTypes.EncodableString -import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext -class HuggingFaceTextSummarizationOpDesc extends PythonOperatorDescriptor { +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.{ + PythonTemplateBuilderStringContext, + pyStringLiteral +} +// type constraint: the tokenizer summarizes text and refuses anything that is not +// a string, so the column can only be a string. +@JsonSchemaInject(json = """ +{ + "attributeTypeRules": { + "attribute": { "enum": ["string"] } + } +} +""") +class HuggingFaceTextSummarizationOpDesc + extends PythonOperatorDescriptor + with StandaloneCodeGenerator { @JsonProperty(value = "attribute", required = true) @JsonPropertyDescription("attribute to perform text summarization on") @AutofillAttributeName + @SampleColumn("long_text") var attribute: EncodableString = _ @JsonProperty( @@ -76,6 +92,37 @@ class HuggingFaceTextSummarizationOpDesc extends PythonOperatorDescriptor { | yield tuple_""".encode } + override def producesDataFrame(): Boolean = true + + // Standalone mirror of generatePythonCode: load the encoder-decoder model + // once, generate a summary per row, and add the STRING result column to + // produce out1df. + override def generateStandaloneCode(): String = { + s"""from transformers import BertTokenizerFast, EncoderDecoderModel + |import torch + | + |model_name = "mrm8488/bert-mini2bert-mini-finetuned-cnn_daily_mail-summarization" + |device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') + |tokenizer = BertTokenizerFast.from_pretrained(model_name) + |model = EncoderDecoderModel.from_pretrained(model_name).to(device) + | + |out1df = in1df.copy() + |_summaries = [] + |for _text in out1df[${pyStringLiteral(attribute)}]: + | # An empty cell arrives as None, which the tokenizer rejects. Keep the row + | # and leave the summary empty rather than ending the run over a value the + | # model has nothing to say about. + | if _text is None or (isinstance(_text, str) and not _text.strip()): + | _summaries.append(None) + | continue + | inputs = tokenizer([_text], padding="max_length", truncation=True, max_length=512, return_tensors="pt") + | input_ids = inputs.input_ids.to(device) + | attention_mask = inputs.attention_mask.to(device) + | output = model.generate(input_ids, attention_mask=attention_mask) + | _summaries.append(tokenizer.decode(output[0], skip_special_tokens=True)) + |out1df[${pyStringLiteral(resultAttribute)}] = _summaries""".stripMargin + } + override def operatorInfo: OperatorInfo = OperatorInfo( "Hugging Face Text Summarization", 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 a2323e5d19f..b21c6dfaa31 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 @@ -19,11 +19,13 @@ package org.apache.texera.amber.operator.sklearn +import org.apache.texera.amber.operator.StandaloneCodeGenerator import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} -abstract class SklearnClassifierOpDesc extends SklearnModelOpDesc { +abstract class SklearnClassifierOpDesc extends SklearnModelOpDesc with StandaloneCodeGenerator { override def getImportStatements = "" @@ -76,4 +78,46 @@ $reportMissingKept ), outputPorts = List(OutputPort(blocking = true)) ) + + override def generateStandaloneCode(): String = { + val estimator = getImportStatements.split(" ").last + val tfidfPart = if (tfidfTransformer) "TfidfTransformer()," else "" + val targetLit = pyStringLiteral(target) + val modelNameLit = pyStringLiteral(getUserFriendlyModelName) + val narrowTrain = dropNonFeatureColumns("X_train", "") + val narrowTest = dropNonFeatureColumns("X_test", "") + + s"""${getImportStatements} + |from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score + |from sklearn.pipeline import make_pipeline + |from sklearn.compose import ColumnTransformer + |from sklearn.feature_extraction.text import CountVectorizer, TfidfTransformer + |import numpy as np + |import pandas as pd + | + |# The same rows the operator drops, and on both frames: the model is fitted + |# on one and scored on the other, so narrowing only one side would fit and + |# score on different data. A local name rather than a reassignment, since + |# the input variable belongs to whichever operator produced it. + |_train = ${dropMissingRowsStandalone("in1df")} + |if len(_train) < len(in1df): + | print("Skipped", len(in1df) - len(_train), "of", len(in1df), "rows with missing values") + |Y_train = _train[$targetLit] + |X_train = _train.drop($targetLit, axis=1) + |$narrowTrain + |model = make_pipeline(${vectorizerStage(c => pyStringLiteral(c))}$tfidfPart$estimator()).fit(X_train, Y_train) + | + |_test = ${dropMissingRowsStandalone("in2df")} + |Y_test = _test[$targetLit] + |X_test = _test.drop($targetLit, axis=1) + |$narrowTest + |predictions = model.predict(X_test) + |print("Overall Accuracy:", round(accuracy_score(Y_test, predictions), 4)) + |f1s = f1_score(Y_test, predictions, average=None) + |precisions = precision_score(Y_test, predictions, average=None) + |recalls = recall_score(Y_test, predictions, average=None) + |for i, class_name in enumerate(np.unique(Y_test)): + | print("Class", repr(class_name), " - F1:", round(f1s[i], 4), ", Precision:", round(precisions[i], 4), ", Recall:", round(recalls[i], 4)) + |out1df = pd.DataFrame([{"model_name": $modelNameLit, "model": model}])""".stripMargin + } } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnLinearRegressionOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnLinearRegressionOpDesc.scala index 9ea793338e1..3bc6154e437 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnLinearRegressionOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnLinearRegressionOpDesc.scala @@ -25,16 +25,23 @@ import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext import org.apache.texera.amber.pybuilder.PyStringTypes.EncodableString import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity} -import org.apache.texera.amber.operator.PythonOperatorDescriptor -import org.apache.texera.amber.operator.metadata.annotations.AutofillAttributeName +import org.apache.texera.amber.operator.{PythonOperatorDescriptor, StandaloneCodeGenerator} +import org.apache.texera.amber.operator.metadata.annotations.{AutofillAttributeName, SampleColumn} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral -class SklearnLinearRegressionOpDesc extends PythonOperatorDescriptor with SklearnFittableColumns { +class SklearnLinearRegressionOpDesc + extends PythonOperatorDescriptor + with StandaloneCodeGenerator + with SklearnFittableColumns { @JsonSchemaTitle("Target Attribute") @JsonPropertyDescription("Attribute in your dataset corresponding to target.") @JsonProperty(required = true) @AutofillAttributeName + // The label the estimator fits against. Test-only steering: without it the + // first column wins, which on a feature/label table is a feature. + @SampleColumn("species") var target: EncodableString = _ @JsonSchemaTitle("Degree") @@ -95,4 +102,39 @@ class SklearnLinearRegressionOpDesc extends PythonOperatorDescriptor with Sklear ) } + override def generateStandaloneCode(): String = { + val targetLit = pyStringLiteral(target) + s"""from sklearn.metrics import mean_absolute_error, r2_score + |from sklearn.pipeline import make_pipeline + |from sklearn.linear_model import LinearRegression + |from sklearn.preprocessing import PolynomialFeatures + |import pandas as pd + | + |# The same rows the operator drops, and on both frames: the model is fitted + |# on one and scored on the other, so narrowing only one side would fit and + |# score on different data. Local names rather than reassignments, since the + |# input variables belong to whichever operators produced them. + |_train = in1df.dropna() + |if len(_train) < len(in1df): + | print("Skipped", len(in1df) - len(_train), "of", len(in1df), "rows with missing values") + |Y_train = _train[$targetLit] + |X_train = _train.drop($targetLit, axis=1) + |${narrowToFittableColumns("X_train", "")} + |pipeline = make_pipeline( + | PolynomialFeatures(degree=$degree), + | LinearRegression() + |) + |model = pipeline.fit(X_train, Y_train) + | + |_test = in2df.dropna() + |Y_test = _test[$targetLit] + |X_test = _test.drop($targetLit, axis=1) + |${narrowToFittableColumns("X_test", "")} + |predictions = model.predict(X_test) + |mae = round(mean_absolute_error(Y_test, predictions), 4) + |r2 = round(r2_score(Y_test, predictions), 4) + |print("MAE:", mae, ", R2:", r2) + |out1df = pd.DataFrame([{"model_name": "LinearRegression", "model": model}])""".stripMargin + } + } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnModelOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnModelOpDesc.scala index c771d31affa..43dd741c0e4 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnModelOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnModelOpDesc.scala @@ -34,12 +34,14 @@ import com.kjetland.jackson.jsonSchema.annotations.{ import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.pybuilder.PyStringTypes.EncodableString import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.PythonOperatorDescriptor import org.apache.texera.amber.operator.metadata.annotations.{ AutofillAttributeName, CommonOpDescAnnotation, - HideAnnotation + HideAnnotation, + SampleColumn } // `text` names the columns Count Vectorizer tokenizes, so they are string columns @@ -72,6 +74,9 @@ import org.apache.texera.amber.operator.metadata.annotations.{ """) abstract class SklearnModelOpDesc extends PythonOperatorDescriptor with SklearnFittableColumns { + // The label the estimator fits against. Test-only steering: without it the + // first column wins, which on a feature/label table is a feature. + @SampleColumn("species") @JsonSchemaTitle("Target Attribute") @JsonPropertyDescription("Attribute in your dataset corresponding to target.") @JsonProperty(required = true) @@ -182,6 +187,22 @@ abstract class SklearnModelOpDesc extends PythonOperatorDescriptor with SklearnF else if (handlesMissingValues) pyb"table.dropna(subset=[$target])".toString else "table.dropna()" + /** [[dropMissingRows]] for the standalone path, which names its own frame and + * has no `self` to decode a column name through. Stated separately rather than + * shared through a renderer: `pyb` encodes by the STATIC type of what it + * interpolates, so a column handed through a `String => String` loses the + * annotation that makes it encode and reaches the script raw. + */ + @JsonIgnore + protected def dropMissingRowsStandalone(frame: String): String = { + val subset = + if (countVectorizer) (text :+ target).map(c => pyStringLiteral(c).toString) + else if (handlesMissingValues) Seq(pyStringLiteral(target).toString) + else Seq.empty + if (subset.isEmpty) s"$frame.dropna()" + else subset.mkString(s"$frame.dropna(subset=[", ", ", "])") + } + // Rows the estimator keeps are still rows the user did not know were incomplete, // so say how many reached the fit. Empty for the estimators that dropped them all. @JsonIgnore @@ -202,6 +223,16 @@ abstract class SklearnModelOpDesc extends PythonOperatorDescriptor with SklearnF s" produces. Turn Count Vectorizer off, or use $alternatives." ) } + // The generated code drops the target before the text pipeline reads its + // columns, so naming it here asks the pipeline for a column that is no + // longer there. Refused rather than vectorized: the label is the answer, + // and a model given it as a feature reads that answer off its own input. + if (text.contains(target)) { + throw new RuntimeException( + s""""$target" is the Target Attribute, so it cannot also be a Text Attribute.""" + + " Remove it from Text Attribute, or fit against a different column." + ) + } } Map( operatorInfo.outputPorts.head.id -> Schema() 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 5d8b1b1e7dc..969186c99b2 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 @@ -24,14 +24,15 @@ import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext import org.apache.texera.amber.pybuilder.PyStringTypes.EncodableString import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity} -import org.apache.texera.amber.operator.PythonOperatorDescriptor +import org.apache.texera.amber.operator.{PythonOperatorDescriptor, StandaloneCodeGenerator} import org.apache.texera.amber.operator.metadata.annotations.{ AutofillAttributeName, AutofillAttributeNameOnPort1 } import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral -class SklearnPredictionOpDesc extends PythonOperatorDescriptor { +class SklearnPredictionOpDesc extends PythonOperatorDescriptor with StandaloneCodeGenerator { @JsonProperty(value = "Model Attribute", required = true, defaultValue = "model") @JsonPropertyDescription("attribute corresponding to ML model") @AutofillAttributeName @@ -99,4 +100,39 @@ class SklearnPredictionOpDesc extends PythonOperatorDescriptor { .add(resultAttribute, resultType) ) } + + /** Python that narrows `X` to the columns the model was fitted on. + * + * The fitting side leaves out the columns an estimator cannot fit, so this + * side has to leave out the same ones or scikit-learn refuses the frame for + * naming features it never saw. Read off the model rather than re-derived: + * what it was fitted on is a fact it carries, and asking it cannot drift from + * whatever rule the fitting operator applied. + */ + private val narrowToFittedFeatures: String = + """_fitted = getattr(model, "feature_names_in_", None) + |if _fitted is not None: + | X = X[list(_fitted)]""".stripMargin + + override def generateStandaloneCode(): String = { + val modelLit = pyStringLiteral(model) + val resultLit = pyStringLiteral(resultAttribute) + if (groundTruthAttribute.nonEmpty) { + s"""from sklearn.pipeline import Pipeline + | + |model = in1df[$modelLit].iloc[0] + |out1df = in2df.copy() + |X = in2df.drop(${pyStringLiteral(groundTruthAttribute)}, axis=1) + |$narrowToFittedFeatures + |out1df[$resultLit] = model.predict(X)""".stripMargin + } else { + s"""from sklearn.pipeline import Pipeline + | + |model = in1df[$modelLit].iloc[0] + |out1df = in2df.copy() + |X = in2df + |$narrowToFittedFeatures + |out1df[$resultLit] = [str(p) for p in model.predict(X)]""".stripMargin + } + } } 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 94319f09367..ea436d05d6b 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 @@ -23,7 +23,7 @@ import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity} -import org.apache.texera.amber.operator.PythonOperatorDescriptor +import org.apache.texera.amber.operator.{PythonOperatorDescriptor, StandaloneCodeGenerator} import org.apache.texera.amber.operator.sklearn.SklearnFittableColumns import org.apache.texera.amber.operator.metadata.annotations.{ AutofillAttributeName, @@ -31,9 +31,15 @@ import org.apache.texera.amber.operator.metadata.annotations.{ } import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} import org.apache.texera.amber.pybuilder.PyStringTypes.EncodableString -import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.{ + PythonTemplateBuilderStringContext, + pyStringLiteral +} -class SklearnTestingOpDesc extends PythonOperatorDescriptor with SklearnFittableColumns { +class SklearnTestingOpDesc + extends PythonOperatorDescriptor + with StandaloneCodeGenerator + with SklearnFittableColumns { @JsonProperty(required = true, defaultValue = "false") @JsonSchemaTitle("Regression") @JsonPropertyDescription( @@ -118,4 +124,28 @@ class SklearnTestingOpDesc extends PythonOperatorDescriptor with SklearnFittable _.add(_, AttributeType.DOUBLE) ) ) + + override def generateStandaloneCode(): String = { + val isRegressionStr = if (isRegression) "True" else "False" + val modelLit = pyStringLiteral(model) + val targetLit = pyStringLiteral(target) + s"""from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score, root_mean_squared_error, mean_absolute_error, r2_score + | + |model = in1df[$modelLit].iloc[0] + |out1df = in1df.copy() + |Y = in2df[$targetLit] + |X = in2df.drop($targetLit, axis=1) + |${narrowToFittableColumns("X", "")} + |predictions = model.predict(X.squeeze()) + |if $isRegressionStr: + | out1df["R2"] = r2_score(Y, predictions) + | out1df["RMSE"] = root_mean_squared_error(Y, predictions) + | out1df["MAE"] = mean_absolute_error(Y, predictions) + |else: + | out1df["accuracy"] = round(accuracy_score(Y, predictions), 4) + | out1df["f1"] = f1_score(Y, predictions, average="weighted") + | out1df["precision"] = precision_score(Y, predictions, average="weighted") + | out1df["recall"] = recall_score(Y, predictions, average="weighted")""".stripMargin + } + } 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 c51f322a628..9dab7c27b56 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 @@ -19,12 +19,14 @@ package org.apache.texera.amber.operator.sklearn.training +import org.apache.texera.amber.operator.StandaloneCodeGenerator import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PortIdentity} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} import org.apache.texera.amber.operator.sklearn.SklearnModelOpDesc -class SklearnTrainingOpDesc extends SklearnModelOpDesc { +class SklearnTrainingOpDesc extends SklearnModelOpDesc with StandaloneCodeGenerator { override def getImportStatements = "" @@ -64,4 +66,30 @@ $reportMissingKept inputPorts = List(InputPort(PortIdentity(), "training")), outputPorts = List(OutputPort(blocking = true)) ) + + override def generateStandaloneCode(): String = { + val estimator = getImportStatements.split(" ").last + val tfidfPart = if (tfidfTransformer) "TfidfTransformer()," else "" + val targetLit = pyStringLiteral(target) + val modelNameLit = pyStringLiteral(getUserFriendlyModelName) + val narrowX = dropNonFeatureColumns("X", "") + + s"""${getImportStatements} + |from sklearn.pipeline import make_pipeline + |from sklearn.compose import ColumnTransformer + |from sklearn.feature_extraction.text import CountVectorizer, TfidfTransformer + |import pandas as pd + | + |# The same rows the operator drops. A local name rather than a + |# reassignment, since the input variable belongs to whichever operator + |# produced it. + |_train = ${dropMissingRowsStandalone("in1df")} + |if len(_train) < len(in1df): + | print("Skipped", len(in1df) - len(_train), "of", len(in1df), "rows with missing values") + |Y = _train[$targetLit] + |X = _train.drop($targetLit, axis=1) + |$narrowX + |model = make_pipeline(${vectorizerStage(c => pyStringLiteral(c))}$tfidfPart$estimator()).fit(X, Y) + |out1df = pd.DataFrame([{"model_name": $modelNameLit, "model": model}])""".stripMargin + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceIrisLogisticRegressionOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceIrisLogisticRegressionOpDescSpec.scala index 1c3701b4d66..730be48ef50 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceIrisLogisticRegressionOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceIrisLogisticRegressionOpDescSpec.scala @@ -19,6 +19,7 @@ package org.apache.texera.amber.operator.huggingFace +import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInject import org.apache.texera.amber.core.executor.OpExecWithCode import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} @@ -105,6 +106,16 @@ class HuggingFaceIrisLogisticRegressionOpDescSpec extends AnyFlatSpec with Match carries(code, "species") shouldBe true } + // An empty cell reaches the standardization as a None, which numpy cannot + // subtract from, so the row is answered rather than ending the run. + it should "leave the prediction empty when a measurement is missing" in { + val d = configured() + val code = d.generatePythonCode() + code should include("if length is None or width is None:") + code should include("yield tuple_") + code should include("return") + } + "HuggingFaceIrisLogisticRegressionOpDesc.getPhysicalOp" should "wire an OpExecWithCode python executor carrying the operator's ports" in { val d = configured() @@ -128,4 +139,18 @@ class HuggingFaceIrisLogisticRegressionOpDescSpec extends AnyFlatSpec with Match h.predictionClassName shouldBe "species" h.predictionProbabilityName shouldBe "probability" } + + "HuggingFaceIrisLogisticRegressionOpDesc (class-level)" should + "carry @JsonSchemaInject restricting both petal columns to numeric attributes" in { + val ann = + classOf[HuggingFaceIrisLogisticRegressionOpDesc].getAnnotation(classOf[JsonSchemaInject]) + ann should not be null + val payload = ann.json + payload should include("attributeTypeRules") + payload should include("petalLengthCmAttribute") + payload should include("petalWidthCmAttribute") + payload should include("integer") + payload should include("long") + payload should include("double") + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSentimentAnalysisOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSentimentAnalysisOpDescSpec.scala index 65401d58322..cdcf5de139c 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSentimentAnalysisOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSentimentAnalysisOpDescSpec.scala @@ -19,6 +19,7 @@ package org.apache.texera.amber.operator.huggingFace +import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInject import org.apache.texera.amber.core.executor.OpExecWithCode import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} @@ -212,4 +213,14 @@ class HuggingFaceSentimentAnalysisOpDescSpec extends AnyFlatSpec with Matchers { h.resultAttributeNeutral shouldBe "neu" h.resultAttributeNegative shouldBe "neg" } + + "HuggingFaceSentimentAnalysisOpDesc (class-level)" should + "carry @JsonSchemaInject restricting `attribute` to STRING columns" in { + val ann = classOf[HuggingFaceSentimentAnalysisOpDesc].getAnnotation(classOf[JsonSchemaInject]) + ann should not be null + val payload = ann.json + payload should include("attributeTypeRules") + payload should include("attribute") + payload should include("string") + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSpamSMSDetectionOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSpamSMSDetectionOpDescSpec.scala index f7d39d7beb5..e13d7e2fd41 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSpamSMSDetectionOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceSpamSMSDetectionOpDescSpec.scala @@ -19,6 +19,7 @@ package org.apache.texera.amber.operator.huggingFace +import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInject import org.apache.texera.amber.core.executor.OpExecWithCode import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} @@ -131,4 +132,14 @@ class HuggingFaceSpamSMSDetectionOpDescSpec extends AnyFlatSpec with Matchers { h.resultAttributeSpam shouldBe "is_spam" h.resultAttributeProbability shouldBe "score" } + + "HuggingFaceSpamSMSDetectionOpDesc (class-level)" should + "carry @JsonSchemaInject restricting `attribute` to STRING columns" in { + val ann = classOf[HuggingFaceSpamSMSDetectionOpDesc].getAnnotation(classOf[JsonSchemaInject]) + ann should not be null + val payload = ann.json + payload should include("attributeTypeRules") + payload should include("attribute") + payload should include("string") + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceTextSummarizationOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceTextSummarizationOpDescSpec.scala index 1c4f3e0d8ba..03bb5e44f66 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceTextSummarizationOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceTextSummarizationOpDescSpec.scala @@ -19,6 +19,7 @@ package org.apache.texera.amber.operator.huggingFace +import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaInject import org.apache.texera.amber.core.executor.OpExecWithCode import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} @@ -135,4 +136,14 @@ class HuggingFaceTextSummarizationOpDescSpec extends AnyFlatSpec with Matchers { h.attribute shouldBe "text" h.resultAttribute shouldBe "summary" } + + "HuggingFaceTextSummarizationOpDesc (class-level)" should + "carry @JsonSchemaInject restricting `attribute` to STRING columns" in { + val ann = classOf[HuggingFaceTextSummarizationOpDesc].getAnnotation(classOf[JsonSchemaInject]) + ann should not be null + val payload = ann.json + payload should include("attributeTypeRules") + payload should include("attribute") + payload should include("string") + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnLinearRegressionOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnLinearRegressionOpDescSpec.scala index d0364e19985..92a0ebac6e9 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnLinearRegressionOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnLinearRegressionOpDescSpec.scala @@ -95,6 +95,19 @@ class SklearnLinearRegressionOpDescSpec extends AnyFlatSpec with Matchers { code should include("No column left to fit on") } + // Both frames, since the model is fitted on one and scored on the other: a + // narrowing on only one side hands `predict` a different feature set. + "SklearnLinearRegressionOpDesc.generateStandaloneCode" should + "narrow both the training and the testing features" in { + val d = new SklearnLinearRegressionOpDesc + d.target = "y" + val code = d.generateStandaloneCode() + code should include("""_fittable = X_train.select_dtypes(include=["number", "bool"])""") + code should include("X_train = _fittable") + code should include("""_fittable = X_test.select_dtypes(include=["number", "bool"])""") + code should include("X_test = _fittable") + } + "SklearnLinearRegressionOpDesc" should "round-trip its target through the polymorphic base" in { val d = new SklearnLinearRegressionOpDesc diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnModelOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnModelOpDescSpec.scala index 127ddabaec6..1ab28a78fb9 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnModelOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnModelOpDescSpec.scala @@ -142,4 +142,32 @@ class SklearnModelOpDescSpec extends AnyFlatSpec with Matchers { } d.getOutputSchemas(Map.empty).keySet shouldBe Set(d.operatorInfo.outputPorts.head.id) } + + it should "reject the target named as a text column" in { + val d = new TestSklearnModelOpDesc + d.countVectorizer = true + d.target = "species" + d.text = List("note", "species") + val thrown = intercept[RuntimeException](d.getOutputSchemas(Map.empty)) + thrown.getMessage should include("species") + thrown.getMessage should include("Target Attribute") + thrown.getMessage should include("Text Attribute") + } + + it should "let the text columns through while none of them is the target" in { + val d = new TestSklearnModelOpDesc + d.countVectorizer = true + d.target = "species" + d.text = List("note") + d.getOutputSchemas(Map.empty).keySet shouldBe Set(d.operatorInfo.outputPorts.head.id) + } + + it should "leave a stale text column alone while Count Vectorizer is off" in { + // Nothing reads `text` with the switch off, and the panel hides it, so a value + // left behind by an earlier configuration must not report the operator invalid. + val d = new TestSklearnModelOpDesc + d.target = "species" + d.text = List("species") + d.getOutputSchemas(Map.empty).keySet shouldBe Set(d.operatorInfo.outputPorts.head.id) + } } 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 16aec5da13d..5c7cf9b5167 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 @@ -148,10 +148,25 @@ class SklearnPredictionOpDescSpec extends AnyFlatSpec with Matchers { d.model = "model" d.resultAttribute = "prediction" d.groundTruthAttribute = "y" - val code = d.generatePythonCode() + Seq(d.generatePythonCode(), d.generateStandaloneCode()).foreach { code => + code should include(""""feature_names_in_", None)""") + code should include("if _fitted is not None:") + } + // The two paths narrow with different expressions: this one holds a Tuple, + // the standalone one a frame. + d.generatePythonCode() should include("input_features.get_partial_tuple(list(_fitted))") + } + + // The branch that names no ground truth predicts on the whole frame, so it needs + // the same narrowing as the one that drops a column first. + it should "narrow the features with no ground-truth column configured too" in { + val d = new SklearnPredictionOpDesc + d.model = "model" + d.resultAttribute = "prediction" + d.groundTruthAttribute = "" + val code = d.generateStandaloneCode() code should include(""""feature_names_in_", None)""") - code should include("if _fitted is not None:") - code should include("input_features.get_partial_tuple(list(_fitted))") + code should include("X = X[list(_fitted)]") } "SklearnPredictionOpDesc" should 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 8401c6e20b8..52b2ad72c8c 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 @@ -70,6 +70,20 @@ class SklearnTestingOpDescSpec extends AnyFlatSpec with Matchers { schema.getAttribute("MAE").getType shouldBe AttributeType.DOUBLE } + // The scorer reads every column but the target, so it has to leave out what an + // estimator cannot fit for the same reason the fitting operators do, and leave + // out the same columns: a model fitted without them refuses a frame naming them. + "SklearnTestingOpDesc" should "narrow the features to the columns an estimator can fit" in { + val d = new SklearnTestingOpDesc + d.model = "model" + d.target = "y" + Seq(d.generatePythonCode(), d.generateStandaloneCode()).foreach { code => + code should include("""_fittable = X.select_dtypes(include=["number", "bool"])""") + code should include("""print("Ignoring columns an estimator cannot fit:", _ignored)""") + code should include("X = _fittable") + } + } + "SklearnTestingOpDesc.generatePythonCode" should "emit the scorer tuple operator" in { val d = new SklearnTestingOpDesc d.model = "model" From abf2c352e0f1b5bdda6b5df543b37e16b366d12c Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 4 Sep 2026 13:29:24 -0700 Subject: [PATCH 2/2] fix(sklearn): score the rows the executor scores The executor drops every row holding a missing value before it scores, because the model arrives already fitted and the operator cannot ask which estimator it holds. The exported script scored all of them, so it answered with a number the run never reported, or failed inside scikit-learn on a value the estimator refuses. The verification runner cannot catch this one: a fixture written from the JVM cannot carry a fitted model on an input port, so the two paths are never run side by side. The test instead pins the property that follows from the drop, which is testable on its own: a row the executor would drop must not move the score. Its label is one the model never saw, so scoring that row is wrong whatever the estimator answers; a plausible label would have agreed by luck, since a decision tree predicts through a missing feature rather than refusing it. Co-Authored-By: Claude Opus 5 (1M context) --- .../testing/SklearnTestingOpDesc.scala | 12 ++- .../testing/SklearnTestingOpDescSpec.scala | 95 +++++++++++++++++++ 2 files changed, 105 insertions(+), 2 deletions(-) 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 ea436d05d6b..cb5e2214192 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 @@ -133,8 +133,16 @@ class SklearnTestingOpDesc | |model = in1df[$modelLit].iloc[0] |out1df = in1df.copy() - |Y = in2df[$targetLit] - |X = in2df.drop($targetLit, axis=1) + |# The same drop the executor makes before it scores: the model arrives + |# already fitted, so this operator cannot ask which estimator it holds and + |# drops on every column to be safe. Scoring the rows it skips would answer + |# with a number the run never reported, or fail inside scikit-learn. + |rows_read = len(in2df) + |scored_df = in2df.dropna() + |if len(scored_df) < rows_read: + | print("Skipped", rows_read - len(scored_df), "of", rows_read, "rows with missing values") + |Y = scored_df[$targetLit] + |X = scored_df.drop($targetLit, axis=1) |${narrowToFittableColumns("X", "")} |predictions = model.predict(X.squeeze()) |if $isRegressionStr: 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 52b2ad72c8c..0f41ee51d7e 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 @@ -19,6 +19,7 @@ package org.apache.texera.amber.operator.sklearn.testing +import com.typesafe.config.ConfigFactory import org.apache.texera.amber.core.tuple.{AttributeType, Schema} import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.LogicalOp @@ -27,6 +28,12 @@ import org.apache.texera.amber.util.JSONUtils.objectMapper import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers +import java.nio.charset.StandardCharsets +import java.nio.file.Files +import java.util.concurrent.TimeUnit +import scala.io.Source +import scala.util.Try + class SklearnTestingOpDescSpec extends AnyFlatSpec with Matchers { "SklearnTestingOpDesc.operatorInfo" should @@ -131,4 +138,92 @@ class SklearnTestingOpDescSpec extends AnyFlatSpec with Matchers { r.model shouldBe "m" r.target shouldBe "t" } + + // The parity this operator cannot get from the verification runner: a fixture + // written from the JVM cannot carry a fitted model on an input port, so the + // two paths are never run side by side. The executor drops the rows holding a + // missing value before it scores, and the property that follows is testable + // here on its own: a row it would drop must not move the score. + it should "score a table with a missing row the way it scores that table without it" in { + val python = resolvePython().getOrElse(cancel("No runnable python executable")) + if (!canImport(python, "pandas, sklearn")) cancel(s"'$python' cannot import pandas and sklearn") + + val op = new SklearnTestingOpDesc + op.model = "model" + op.target = "target" + // The generated block reads in1df / in2df and writes out1df, so it runs as + // the body of the loop below. + val body = op.generateStandaloneCode().linesIterator.map(" " + _).mkString("\n") + + val driver = + s"""import pandas as pd + |from sklearn.tree import DecisionTreeClassifier + | + |train = pd.DataFrame({"f1": [0, 1, 0, 1], "f2": [0, 0, 1, 1], "target": [0, 1, 1, 0]}) + |fitted = DecisionTreeClassifier(random_state=0).fit(train[["f1", "f2"]], train["target"]) + | + |# The same rows twice, except that one carries a hole the executor would + |# drop. Reading the same score from both is the parity. + |# + |# That row's label is one the model was never trained on, so scoring it + |# is wrong whatever the estimator answers. Without it the score has to + |# move, which is what makes this test able to fail: a decision tree + |# predicts through a missing feature rather than refusing it, so a row + |# with a plausible label would have agreed by luck. + |with_hole = pd.DataFrame( + | {"f1": [0.0, 1.0, 0.0, None], "f2": [0.0, 0.0, 1.0, 1.0], "target": [0, 1, 1, 2]} + |) + |without = with_hole.dropna().reset_index(drop=True) + | + |scores = [] + |for frame in (with_hole, without): + | in1df = pd.DataFrame({"model": [fitted]}) + | in2df = frame + |$body + | scores.append(round(float(out1df["accuracy"].iloc[0]), 10)) + | + |print(scores[0]) + |print(scores[1]) + |""".stripMargin + + val script = Files.createTempFile("sklearn-testing-parity-", ".py") + script.toFile.deleteOnExit() + Files.write(script, driver.getBytes(StandardCharsets.UTF_8)) + val process = new ProcessBuilder(python, script.toString).redirectErrorStream(true).start() + val out = Source.fromInputStream(process.getInputStream).mkString + process.waitFor(180, TimeUnit.SECONDS) + withClue(s"python said:\n$out\nscript:\n$driver") { process.exitValue() shouldBe 0 } + + val answers = out.trim.linesIterator.filter(_.matches("""-?\d+\.\d+""")).toSeq + withClue(s"python said:\n$out") { + answers should have length 2 + answers.head shouldBe answers(1) + } + } + + private def resolvePython(): Option[String] = { + def fromConfig: Option[String] = + Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption + .orElse(Try(ConfigFactory.load()).toOption) + .flatMap(c => Try(c.getConfig("python").getString("path")).toOption) + .map(_.trim) + .filter(_.nonEmpty) + + def runnable(exe: String): Boolean = + Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption + .exists { p => + if (!p.waitFor(5, TimeUnit.SECONDS)) { p.destroyForcibly(); false } + else p.exitValue() == 0 + } + + (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(runnable) + } + + private def canImport(python: String, modules: String): Boolean = + Try( + new ProcessBuilder(python, "-c", s"import $modules").redirectErrorStream(true).start() + ).toOption.exists { p => + if (!p.waitFor(120, TimeUnit.SECONDS)) { p.destroyForcibly(); false } + else p.exitValue() == 0 + } }