Sail examples

Appearance

Python UDFsView Python source

A filter can make a Python function run again

As in Spark, a deterministic Python UDF can run more than once per row when the query filters on its output. Mark UDFs that are expensive or have side effects (model or API calls) with .asNondeterministic(), which makes them run once per row.

Check targets: pysail==0.7.3 with pyspark-client==4.1.3

The Python assertions below check this behavior. Details attributed to external documentation are outside those assertions. Newer releases may behave differently.

Run the check

Follow the test setup, then run from the repository root (sail-examples/). The harness applies the configuration declared in this check.

Sail

uv run pytest 'tests/test_facts.py::test_fact[udf-reruns-when-filtered[sail]]'

Apache Spark (requires Java 17)

uv run pytest 'tests/test_facts.py::test_fact[udf-reruns-when-filtered[spark]]'

Python code

udf-reruns-when-filtered.py40 lines
View udf-reruns-when-filtered.py on GitHub 
FACT = {
    "id": "udf-reruns-when-filtered",
    "title": "A filter can make a Python function run again",
    "category": "python-udfs",
    "text": """As in Spark, a deterministic Python UDF can run more than once per row when the query filters on its
        output. Mark UDFs that are expensive or have side effects (model or API calls) with
        `.asNondeterministic()`, which makes them run once per row.""",
    "engines": ["sail", "spark"],
    "order": 3,
}

import tempfile
from pathlib import Path

from pyspark.sql import functions as F


def calls_for(spark, nondeterministic: bool, rows: int) -> int:
    log = Path(tempfile.mkdtemp()) / "calls.log"
    log_path = str(log)

    def body(x):
        with open(log_path, "a") as f:
            f.write("x\n")
        return x

    udf = F.udf(body, "long")
    if nondeterministic:
        udf = udf.asNondeterministic()
    spark.range(rows).select(udf("id").alias("v")).filter("v >= 0").collect()
    return len(log.read_text().splitlines())


def check(spark, engine):
    rows = 40
    deterministic = calls_for(spark, False, rows)
    nondeterministic = calls_for(spark, True, rows)
    assert deterministic > rows, f"expected re-runs for a deterministic UDF, got {deterministic}"
    assert nondeterministic == rows, f"asNondeterministic() should run once per row, got {nondeterministic}"
    return {"rows": rows, "deterministic_calls": deterministic, "nondeterministic_calls": nondeterministic}