This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8367-6d0931cb9b193d87f477cca039c9c805126f452d in repository https://gitbox.apache.org/repos/asf/texera.git
commit 506397a5c2d8f392a1f8ceb85dc5f7a5416169b8 Author: Kary Zheng <[email protected]> AuthorDate: Mon Sep 28 21:35:20 2026 +0000 feat(machine-learning): export the Machine Learning Scorer (#8367) ### What changes were proposed in this PR? The scorer reads one label through two columns, the actual and the predicted, and reports the metrics a task admits. The export says which task it is scoring the same way the operator does, so the two paths select the same metrics from the same pair. Two behaviour fixes travel with it. It reports the settings it cannot score rather than failing deep inside the generated Python, and it scores the rows an empty cell leaves usable rather than ending the run on one. ### Any related issues, documentation, discussions? Part of #8325, 19 of 27; that issue lists the set in order. It needs #8327 for the trait, so it does not compile until that lands, and the rows these operators add to the verification runner follow with the harness rather than as whole new files here. Closes #8302, closes #8058. Closes #8421, the task this change is the whole of. ### How was this PR tested? Each operator asserts the block it emits in its own spec. Once the harness lands, each is also run through the engine and through its generated script, on every configuration its schema offers, and the two answers compared. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --------- Co-authored-by: Claude Opus 5 (1M context) <[email protected]> --- .../Scorer/MachineLearningScorerOpDesc.scala | 159 +++++++++++++- .../Scorer/MachineLearningScorerOpDescSpec.scala | 238 +++++++++++++++++++++ 2 files changed, 394 insertions(+), 3 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDesc.scala index e43c3f3947..7ad96e3b3c 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDesc.scala @@ -29,11 +29,46 @@ import org.apache.texera.amber.core.tuple.{Attribute, 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, HideAnnotation} +import org.apache.texera.amber.operator.{PythonOperatorDescriptor, StandaloneCodeGenerator} +import org.apache.texera.amber.operator.metadata.annotations.{ + AutofillAttributeName, + HideAnnotation, + SampleColumn +} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral -class MachineLearningScorerOpDesc extends PythonOperatorDescriptor { +// Whichever metric list the task shows must hold at least one metric: an empty +// one renders as `metric_list = ['']` and the metric lookup fails on the empty +// name. The rule is conditional because only one of the two lists is visible at +// a time, and a flat `required` on both can never be satisfied. +@JsonSchemaInject(json = """ +{ + "allOf": [ + { + "if": { + "required": ["isRegression"], + "properties": { "isRegression": { "const": false } } + }, + "then": { + "required": ["classificationFlag"], + "properties": { "classificationFlag": { "minItems": 1 } } + } + }, + { + "if": { + "required": ["isRegression"], + "properties": { "isRegression": { "const": true } } + }, + "then": { + "required": ["regressionFlag"], + "properties": { "regressionFlag": { "minItems": 1 } } + } + } + ] +} +""") +class MachineLearningScorerOpDesc extends PythonOperatorDescriptor with StandaloneCodeGenerator { @JsonProperty(required = true, defaultValue = "false") @JsonSchemaTitle("Regression") @JsonPropertyDescription( @@ -45,12 +80,16 @@ class MachineLearningScorerOpDesc extends PythonOperatorDescriptor { @JsonSchemaTitle("Actual Value") @JsonPropertyDescription("Specify the label attribute") @AutofillAttributeName + // Test-only steering: the pair has to be one label read twice, and the + // first-unused pick names two unrelated columns. + @SampleColumn("a\"b\\c_species") var actualValueColumn: EncodableString = "" @JsonProperty(required = true) @JsonSchemaTitle("Predicted Value") @JsonPropertyDescription("Specify the attribute generated by the model") @AutofillAttributeName + @SampleColumn("a\"b\\c_species_pred") var predictValueColumn: EncodableString = "" @JsonProperty(required = false, value = "classificationFlag") @@ -85,9 +124,61 @@ class MachineLearningScorerOpDesc extends PythonOperatorDescriptor { inputPorts = List(InputPort()), outputPorts = List(OutputPort()) ) + + /** The two scored columns hold one quantity measured twice, so they have to be + * comparable. Left to sklearn, a mismatched pair either raises `Mix of label + * input types` from deep inside a metric or, for Accuracy, quietly scores 0 — + * neither of which points back at the two columns the user picked. + * + * Types only, not values: a DOUBLE label column of 0.0 / 1.0 is a legitimate + * classification target even though a continuous one is not, and the schema + * cannot tell the two apart. + */ + private def validateScoredColumns(inputSchemas: Map[PortIdentity, Schema]): Unit = { + val numeric = Set(AttributeType.INTEGER, AttributeType.LONG, AttributeType.DOUBLE) + // Blank names are what the form reports before the user has picked; the + // schema's own `required` already says so, and saying it twice would put a + // second message on a half-filled operator. + if (actualValueColumn.isEmpty || predictValueColumn.isEmpty) return + inputSchemas.get(operatorInfo.inputPorts.head.id).foreach { schema => + Seq(("Actual Value", actualValueColumn), ("Predicted Value", predictValueColumn)) + .foreach { + case (label, column) => + if (!schema.containsAttribute(column)) + throw new RuntimeException(s"$label column '$column' is not in the input table") + } + val actualType = schema.getAttribute(actualValueColumn).getType + val predictType = schema.getAttribute(predictValueColumn).getType + if (isRegression) { + Seq( + ("Actual Value", actualValueColumn, actualType), + ("Predicted Value", predictValueColumn, predictType) + ) + .foreach { + case (label, column, attributeType) => + if (!numeric.contains(attributeType)) + throw new RuntimeException( + s"A regression metric needs a numeric $label column, but '$column' is " + + s"${attributeType.getName}" + ) + } + } else { + val comparable = (numeric.contains(actualType) && numeric.contains(predictType)) || + actualType == predictType + if (!comparable) + throw new RuntimeException( + s"Actual Value '$actualValueColumn' (${actualType.getName}) and Predicted Value " + + s"'$predictValueColumn' (${predictType.getName}) hold different kinds of label, " + + "so a classification metric cannot compare them" + ) + } + } + } + override def getOutputSchemas( inputSchemas: Map[PortIdentity, Schema] ): Map[PortIdentity, Schema] = { + validateScoredColumns(inputSchemas) val metrics = if (isRegression) { regressionMetrics.map(_.getName()) } else { @@ -176,6 +267,14 @@ class MachineLearningScorerOpDesc extends PythonOperatorDescriptor { | | @overrides | def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]: + | # A row missing either value has nothing to score, and the metrics + | # refuse the empty cell rather than passing over it. + | table = table.dropna(subset=[$actualValueColumn, $predictValueColumn]) + | # Nothing survived the drop. The metrics answer that badly and each in + | # its own way: the regression ones raise from inside scikit-learn, and + | # the classification ones return a NaN score, which reads as a result. + | if table.empty: + | raise ValueError("No rows left to score: every row is missing the actual value, the predicted value, or both.") | y_true = table[$actualValueColumn] | y_pred = table[$predictValueColumn] | @@ -193,4 +292,58 @@ class MachineLearningScorerOpDesc extends PythonOperatorDescriptor { finalcode.encode } + override def generateStandaloneCode(): String = { + val isRegressionStr = if (isRegression) "True" else "False" + s"""import pandas as pd + |from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, mean_squared_error, root_mean_squared_error, mean_absolute_error, r2_score + | + |def classification_metrics(y_true, y_pred, metric_list, labels): + | if 'Accuracy' in metric_list: + | labels.insert(0, 'Overall') + | result = {metric: [None] * len(labels) for metric in metric_list} + | metrics_func = {'Precision Score': precision_score, 'Recall Score': recall_score, 'F1 Score': f1_score} + | + | for metric in metric_list: + | if metric == 'Accuracy': + | result['Accuracy'][0] = round(accuracy_score(y_true, y_pred), 4) + | else: + | for i, label in enumerate(labels): + | if label != 'Overall': + | prediction = metrics_func[metric](y_true, y_pred, average=None, labels=[label]) + | result[metric][i] = round(prediction[0], 4) + | + | labels = ['class_' + str(label) if type(label) != str else label for label in labels] + | result['Class'] = labels + | # Class first and the metrics in the order they were picked, which is + | # the column order this operator declares and so the order of the rows + | # the workflow shows. + | return pd.DataFrame(result)[['Class'] + metric_list] + | + |def regression_metrics(y_true, y_pred, metric_list): + | result = dict() + | metrics_func = {'MSE': mean_squared_error, 'RMSE': root_mean_squared_error, 'MAE': mean_absolute_error, 'R2': r2_score} + | for metric in metric_list: + | result[metric] = metrics_func[metric](y_true, y_pred) + | return pd.DataFrame(result, index=[0]) + | + |# Bound to a name of its own: the same frame can feed another branch of the + |# plan, which must still see every row. + |scored_df = in1df.dropna(subset=[${pyStringLiteral(actualValueColumn)}, ${pyStringLiteral( + predictValueColumn + )}]) + |# Nothing survived the drop. The metrics answer that badly and each in its own + |# way: the regression ones raise from inside scikit-learn, and the + |# classification ones return a NaN score, which reads as a result. + |if scored_df.empty: + | raise ValueError("No rows left to score: every row is missing the actual value, the predicted value, or both.") + |y_true = scored_df[${pyStringLiteral(actualValueColumn)}] + |y_pred = scored_df[${pyStringLiteral(predictValueColumn)}] + |metric_list = [${getSelectedMetrics()}] + |if $isRegressionStr: + | out1df = regression_metrics(y_true, y_pred, metric_list) + |else: + | labels = list(set(y_true)) + | out1df = classification_metrics(y_true, y_pred, metric_list, labels)""".stripMargin + } + } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDescSpec.scala index 09f916a51d..c21ff4c439 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDescSpec.scala @@ -20,18 +20,27 @@ package org.apache.texera.amber.operator.machineLearning.Scorer import com.fasterxml.jackson.databind.node.ObjectNode +import com.typesafe.config.ConfigFactory import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema} +import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.LogicalOp import org.apache.texera.amber.operator.metadata.OperatorGroupConstants +import org.apache.texera.amber.operator.tags.IntegrationTest import org.apache.texera.amber.util.JSONUtils.objectMapper +import org.scalatest.Tag import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers import java.nio.charset.StandardCharsets +import java.nio.file.Files import java.util.Base64 +import java.util.concurrent.TimeUnit +import scala.util.Try class MachineLearningScorerOpDescSpec extends AnyFlatSpec with Matchers { + private val NeedsPythonPackages = Tag(classOf[IntegrationTest].getName) + /** An EncodableString field renders as a runtime decode site in the emitted code. */ private val decodeSite = "self.decode_python_template" @@ -90,6 +99,19 @@ class MachineLearningScorerOpDescSpec extends AnyFlatSpec with Matchers { ) } + // `result['Class']` is assigned after the metrics, so the frame comes back with + // Class last, where the schema above declares it first and the engine writes its + // rows through that schema. The exported script has none to write through, so it + // put the same numbers under a different column order. + it should "put Class first in the exported frame, as the schema declares it" in { + val d = new MachineLearningScorerOpDesc + d.classificationMetrics = + List(classificationMetricsFnc.accuracy, classificationMetricsFnc.f1Score) + d.generateStandaloneCode() should include( + "return pd.DataFrame(result)[['Class'] + metric_list]" + ) + } + it should "produce an empty schema for regression with no metrics" in { val d = new MachineLearningScorerOpDesc d.isRegression = true @@ -109,6 +131,67 @@ class MachineLearningScorerOpDescSpec extends AnyFlatSpec with Matchers { List(("MSE", AttributeType.DOUBLE), ("R2", AttributeType.DOUBLE)) } + /** An input table holding one column of each type the scored-column check reasons about. */ + private def inputSchemas(d: MachineLearningScorerOpDesc): Map[PortIdentity, Schema] = + Map( + d.operatorInfo.inputPorts.head.id -> Schema( + List( + new Attribute("y_int", AttributeType.INTEGER), + new Attribute("pred_long", AttributeType.LONG), + new Attribute("label_str", AttributeType.STRING), + new Attribute("pred_str", AttributeType.STRING) + ) + ) + ) + + private def scorer( + regression: Boolean, + actual: String, + predict: String + ): MachineLearningScorerOpDesc = { + val d = new MachineLearningScorerOpDesc + d.isRegression = regression + d.actualValueColumn = actual + d.predictValueColumn = predict + d + } + + it should "accept a classification pair whose types are both numeric" in { + // INTEGER against LONG is one label domain read through two widths, which is + // what an upstream predictor emitting a wider column looks like. + val d = scorer(regression = false, "y_int", "pred_long") + d.classificationMetrics = List(classificationMetricsFnc.accuracy) + noException should be thrownBy d.getOutputSchemas(inputSchemas(d)) + } + + it should "reject a classification pair that mixes a number with a string" in { + val d = scorer(regression = false, "y_int", "pred_str") + d.classificationMetrics = List(classificationMetricsFnc.accuracy) + the[RuntimeException] thrownBy d.getOutputSchemas(inputSchemas(d)) should have message + "Actual Value 'y_int' (integer) and Predicted Value 'pred_str' (string) hold different " + + "kinds of label, so a classification metric cannot compare them" + } + + it should "reject a non-numeric column once the task is regression" in { + val d = scorer(regression = true, "label_str", "pred_str") + d.regressionMetrics = List(regressionMetricsFnc.mse) + the[RuntimeException] thrownBy d.getOutputSchemas(inputSchemas(d)) should have message + "A regression metric needs a numeric Actual Value column, but 'label_str' is string" + } + + it should "reject a column the input table does not hold" in { + val d = scorer(regression = false, "y_int", "absent") + the[RuntimeException] thrownBy d.getOutputSchemas(inputSchemas(d)) should have message + "Predicted Value column 'absent' is not in the input table" + } + + it should "say nothing about the pair while a column is still unpicked" in { + // Half-filled is the state every operator passes through on the way to a + // valid one; the empty field already carries its own required marker. + val d = scorer(regression = false, "y_int", "") + noException should be thrownBy d.getOutputSchemas(inputSchemas(d)) + } + "MachineLearningScorerOpDesc.generatePythonCode" should "emit the scorer table operator" in { val d = new MachineLearningScorerOpDesc d.actualValueColumn = "y" @@ -127,6 +210,22 @@ class MachineLearningScorerOpDescSpec extends AnyFlatSpec with Matchers { code should include("if False:") } + it should "drop the rows missing either scored column, on both paths" in { + // The metrics refuse an empty cell instead of passing over it, so the row has + // to go before it reaches them. Both generators emit the same drop, and the + // subset names both columns: dropping on either one alone would leave the two + // series misaligned. + val d = new MachineLearningScorerOpDesc + d.actualValueColumn = "y" + d.predictValueColumn = "yhat" + d.generatePythonCode() should include( + s"table = table.dropna(subset=[$decodeSite('${b64("y")}'), $decodeSite('${b64("yhat")}')])" + ) + // Each binds the drop to a name of its own rather than writing it back: the + // frame this operator was handed can feed another branch of the plan. + d.generateStandaloneCode() should include("""scored_df = in1df.dropna(subset=["y", "yhat"])""") + } + it should "splice the selected metrics verbatim into a proper metric_list" in { // The metric fragment must be spliced verbatim, not re-encoded as one quoted // value (which would collapse the whole list into a single malformed element). @@ -187,4 +286,143 @@ class MachineLearningScorerOpDescSpec extends AnyFlatSpec with Matchers { s.actualValueColumn shouldBe "y" s.predictValueColumn shouldBe "yhat" } + + it should "refuse a table the drop leaves empty, rather than scoring nothing" in { + // The metrics answer an emptied table badly and each in its own way, so the + // operator has to say what happened before they are reached. + val d = new MachineLearningScorerOpDesc + d.actualValueColumn = "y" + d.predictValueColumn = "yhat" + d.generatePythonCode() should include("if table.empty:") + // The standalone path drops the same rows, so it has to answer the same way. + d.generateStandaloneCode() should include("if scored_df.empty:") + Seq(d.generatePythonCode(), d.generateStandaloneCode()).foreach { code => + code should include("No rows left to score") + } + } + + // Python executable resolution, following FilledAreaPlotOpDescSpec: + // udf.conf python.path (UDF_PYTHON_PATH), then python3 / python / py. + private def resolvePythonExecutable(): Option[String] = { + def fromConfig: Option[String] = { + val configOpt = + Try(ConfigFactory.parseResources("udf.conf").resolve()).toOption + .orElse(Try(ConfigFactory.load()).toOption) + configOpt + .flatMap(c => Try(c.getConfig("python").getString("path")).toOption) + .map(_.trim) + .filter(_.nonEmpty) + } + + def isRunnable(exe: String): Boolean = { + val pTry = Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()) + pTry.toOption.exists { p => + val finished = p.waitFor(5, TimeUnit.SECONDS) + if (!finished) { p.destroyForcibly(); false } + else p.exitValue() == 0 + } + } + + (fromConfig.toList ++ List("python3", "python", "py")).distinct.find(isRunnable) + } + + private def canImportPandasAndSklearn(python: String): Boolean = { + val pTry = Try( + new ProcessBuilder(python, "-c", "import pandas, sklearn").redirectErrorStream(true).start() + ) + pTry.toOption.exists { p => + val finished = p.waitFor(60, TimeUnit.SECONDS) + if (!finished) { p.destroyForcibly(); false } + else p.exitValue() == 0 + } + } + + // Driver executed by the runtime test below. It stubs only the pytexera import seam; + // the generated module runs unmodified, against the real scikit-learn metrics. + private val runtimeDriverScript: String = + """import base64 + |import sys + |import types + |from typing import Iterator, Optional + | + |import numpy as np + |import pandas as pd + | + |class UDFTableOperator: + | def decode_python_template(self, data): + | return base64.b64decode(data).decode("utf-8") + | + |stub = types.ModuleType("pytexera") + |stub.UDFTableOperator = UDFTableOperator + |stub.overrides = lambda fn: fn + |stub.Table = pd.DataFrame + |stub.TableLike = object + |stub.Iterator = Iterator + |stub.Optional = Optional + |sys.modules["pytexera"] = stub + | + |ns = {"__name__": "generated_scorer"} + |with open(sys.argv[1]) as f: + | exec(compile(f.read(), sys.argv[1], "exec"), ns) + |op = ns["ProcessTableOperator"]() + | + |nan = float("nan") + |cases = [ + | ("filled", [(1, 1), (0, 0), (1, 1)]), + | ("some_blank", [(nan, 0), (0, nan), (1, 1), (1, 0), (1, 1)]), + | ("all_blank", [(nan, 0), (0, nan), (nan, nan)]), + |] + | + |for cid, rows in cases: + | frame = pd.DataFrame(rows, columns=["y", "yhat"]) + | try: + | scored = list(op.process_table(frame, 0))[0] + | print("CASE %s ACCURACY %s" % (cid, scored["Accuracy"][0])) + | except ValueError as e: + | print("CASE %s REFUSED %s" % (cid, e)) + |""".stripMargin + + it should "score the rows it can and refuse a table that keeps none of them" taggedAs NeedsPythonPackages in { + val python = resolvePythonExecutable().getOrElse( + cancel("No runnable python executable (udf.conf python.path, python3, python, py)") + ) + if (!canImportPandasAndSklearn(python)) { + cancel(s"'$python' cannot import pandas and sklearn; skipping runtime verification") + } + + val d = new MachineLearningScorerOpDesc + d.actualValueColumn = "y" + d.predictValueColumn = "yhat" + d.classificationMetrics = List(classificationMetricsFnc.accuracy) + + val moduleFile = Files.createTempFile("scorer_op_", ".py") + val driverFile = Files.createTempFile("scorer_driver_", ".py") + try { + Files.write(moduleFile, d.generatePythonCode().getBytes(StandardCharsets.UTF_8)) + Files.write(driverFile, runtimeDriverScript.getBytes(StandardCharsets.UTF_8)) + + val process = new ProcessBuilder(python, driverFile.toString, moduleFile.toString) + .redirectErrorStream(true) + .start() + val finished = process.waitFor(120, TimeUnit.SECONDS) + if (!finished) { + process.destroyForcibly() + fail("Scoring driver timed out after 120s") + } + val output = new String(process.getInputStream.readAllBytes(), StandardCharsets.UTF_8) + withClue(s"Driver output:\n$output\n") { + process.exitValue() shouldBe 0 + // Every row usable: all three agree, so the score is 1. + output should include("CASE filled ACCURACY 1.0") + // Two rows dropped, three scored, two of those correct. + output should include("CASE some_blank ACCURACY 0.6667") + // Nothing survives the drop, and the operator says so rather than + // handing back the NaN the classification metrics would produce. + output should include("CASE all_blank REFUSED No rows left to score") + } + } finally { + Files.deleteIfExists(moduleFile) + Files.deleteIfExists(driverFile) + } + } }
