Sail examples

Appearance

Python UDFsView Python source

Create Python UDFs after starting the session

Create UDFs after the Spark session exists. With PySpark 4 clients, @udf, @pandas_udf and @arrow_udf applied at import time, before any session, fail with NO_ACTIVE_OR_DEFAULT_SESSION (or SESSION_OR_CONTEXT_NOT_EXISTS). Keep plain functions at module level and wrap them, for example F.udf(fn, "double"), once the session is up.

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[udfs-need-a-session[sail]]'

Apache Spark (requires Java 17)

uv run pytest 'tests/test_facts.py::test_fact[udfs-need-a-session[spark]]'

Python code

udfs-need-a-session.py39 lines
View udfs-need-a-session.py on GitHub 
FACT = {
    "id": "udfs-need-a-session",
    "title": "Create Python UDFs after starting the session",
    "category": "python-udfs",
    "text": """Create UDFs after the Spark session exists. With PySpark 4 clients, `@udf`,
        `@pandas_udf` and `@arrow_udf` applied at import time, before any session, fail with
        `NO_ACTIVE_OR_DEFAULT_SESSION` (or `SESSION_OR_CONTEXT_NOT_EXISTS`). Keep plain functions at
        module level and wrap them, for example `F.udf(fn, "double")`, once the session is up.""",
    "engines": ["sail", "spark"],
    "order": 3,
}

import subprocess
import sys

PROBE = """
import pandas as pd
from pyspark.sql import functions as F
from pyspark.sql.functions import pandas_udf
def f(a: pd.Series) -> pd.Series:
    return a
for make in (lambda: F.udf("double")(lambda x: x), lambda: pandas_udf("double")(f)):
    try:
        make()
        print("CREATED")
    except Exception as error:
        print("FAILED", " ".join(str(error).split())[:120])
"""


def check(spark, engine):
    # A fresh interpreter in this engine's environment, with no session in it.
    probe = [sys.executable, "-c", PROBE]
    out = subprocess.run(probe, capture_output=True, text=True, timeout=120, check=False).stdout
    lines = [line for line in out.splitlines() if line.startswith(("CREATED", "FAILED"))]
    assert len(lines) == 2, f"probe did not report: {out[-300:]}"
    assert all(line.startswith("FAILED") for line in lines), f"a UDF was created without a session: {lines}"
    assert all("SESSION" in line for line in lines), f"failed for another reason: {lines}"
    return {"without_session": [line[:80] for line in lines]}