Sail examples

Appearance

Behaves differently from SparkView Python source

Caching does not save the result

cache() and persist() are accepted but do not keep results: every action recomputes the DataFrame, including any UDF or model call in it. To reuse an expensive result (for example model outputs), write it to Parquet or a table and read it back.

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[cache-does-not-keep-results[sail]]'

Apache Spark (requires Java 17)

uv run pytest 'tests/test_facts.py::test_fact[cache-does-not-keep-results[spark]]'

Python code

cache-does-not-keep-results.py44 lines
View cache-does-not-keep-results.py on GitHub 
FACT = {
    "id": "cache-does-not-keep-results",
    "title": "Caching does not save the result",
    "category": "silent-behavior",
    "core": True,
    "text": """`cache()` and `persist()` are accepted but do not keep results: every action recomputes
        the DataFrame, including any UDF or model call in it. To reuse an expensive result (for example
        model outputs), write it to Parquet or a table and read it back.""",
    "engines": ["sail", "spark"],
}

import tempfile
from pathlib import Path

from pyspark.sql import functions as F


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

    @F.udf("long")
    def counted(x):
        with open(log_path, "a") as f:
            f.write("x\n")
        return x

    df = keep(spark.range(rows).select(counted("id").alias("v")))
    df.collect()
    df.collect()
    return len(log.read_text().splitlines())


def check(spark, engine):
    rows = 20
    calls = {
        "cache": udf_calls(spark, lambda df: df.cache(), rows),
        "persist": udf_calls(spark, lambda df: df.persist(), rows),
    }
    expected = 2 * rows if engine == "sail" else rows
    assert calls == {"cache": expected, "persist": expected}, (
        f"expected {expected} UDF calls each, got {calls}"
    )
    return {"rows": rows, "udf_calls": calls}