Choose a supported file format

Parquet, CSV, JSON, text and Delta Lake read and write with the standard spark.read and df.write calls and need no extra packages. Avro can be read but not written. ORC (read or write) and the image source are not available.

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[file-formats[sail]]'

Python code

file-formats.py68 lines

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68# /// script
# dependencies = ["fastavro>=1.9"]
# ///
FACT = {
    "id": "file-formats",
    "title": "Choose a supported file format",
    "category": "data-sources",
    "core": True,
    "text": """Parquet, CSV, JSON, text and Delta Lake read and write with the standard `spark.read`
        and `df.write` calls and need no extra packages. Avro can be read but not written. ORC (read or
        write) and the `image` source are not available.""",
    "engines": ["sail"],
    "order": 2,
}

import tempfile
from pathlib import Path

import fastavro
import pyarrow as pa
from pyarrow import orc


def attempt(action) -> str | None:
    try:
        action()
    except Exception as error:  # noqa: BLE001
        return " ".join(str(error).split())
    return None


def check(spark, engine):
    work = Path(tempfile.mkdtemp())
    df = spark.range(5).selectExpr("id", "cast(id AS string) AS s")
    for fmt in ("parquet", "csv", "json", "delta"):
        path = str(work / fmt)
        df.write.format(fmt).option("header", "true").save(path)
        got = spark.read.format(fmt).option("header", "true").load(path).count()
        assert got == 5, f"{fmt}: wrote 5 rows, read {got}"
    df.select("s").write.text(str(work / "text"))
    assert spark.read.text(str(work / "text")).count() == 5, "text round trip failed"

    (work / "avro-in").mkdir()
    schema = {"type": "record", "name": "r", "fields": [{"name": "id", "type": "long"}]}
    with open(work / "avro-in" / "part.avro", "wb") as f:
        fastavro.writer(f, schema, [{"id": 1}, {"id": 2}, {"id": 3}])
    avro_rows = spark.read.format("avro").load(str(work / "avro-in")).count()
    assert avro_rows == 3, f"Avro read returned {avro_rows} rows, expected 3"

    (work / "orc-in").mkdir()
    orc.write_table(pa.table({"id": [1, 2]}), str(work / "orc-in" / "part.orc"))
    failures = {
        "avro write": attempt(lambda: df.write.format("avro").save(str(work / "avro-out"))),
        "orc read": attempt(lambda: spark.read.orc(str(work / "orc-in")).collect()),
        "orc write": attempt(lambda: df.write.orc(str(work / "orc-out"))),
        "image read": attempt(lambda: spark.read.format("image").load(str(work / "parquet")).collect()),
    }
    now_working = [name for name, error in failures.items() if error is None]
    assert not now_working, f"now available: {now_working}; update this fact"
    # Each error must be about the feature itself; the exact wording may change between releases.
    expected = {"avro write": "not implemented", "orc read": "orc", "orc write": "orc", "image read": "image"}
    wrong = {k: failures[k] for k, text in expected.items() if text not in failures[k].lower()}
    assert not wrong, f"failed for other reasons: {wrong}"
    return {
        "round_trip": ["parquet", "csv", "json", "delta", "text"],
        "avro_read_rows": avro_rows,
        "unavailable": failures,
    }

Choose a supported file format

Run the check

Sail

Python code

Related examples