The Daft engine of Volcano Engine LAS AI provides the Checkpoint feature, enabling resumable computation and incremental computation in production environments for batch data processing. It leverages previously written results for deduplication, filtering out unnecessary computations. This document introduces the details and practical usage of the Checkpoint feature based on the Daft engine.

When using distributed frameworks to preprocess multimodal data (such as images, text, audio, video, and more), the process often fails and terminates due to various reasons, such as resource constraints, unstable environments, or unrobust code. Once a failure occurs during the process, restarting typically means the entire workflow runs from the beginning, causing the already processed data from the previous run to be processed again. This redundant computation results in significant waste of resources and time.
Therefore, based on the Daft engine, Volcano Engine LAS AI provides the Checkpoint feature: enabling resumable computation and incremental computation in production environments for batch data writing. For example, if the first run only processes and writes part of the data before failing, subsequent runs will process only the remaining data, avoiding duplication, ensuring idempotency, and saving cost and time.

Caution and limitations #

Usage #

df = daft.read_parquet(source_dir)

# Intermediate processing flow
df = df.with_column(...)
df = df.with_column(...)

# Configure checkpoint parameters
checkpoint_config = {
    "key_column": "object_key",
}

# Write
df.write_parquet(dest_dir, checkpoint_config=checkpoint_config)

Configure checkpoint #

Description of checkpoint parameters #

Demo 1: Using the primary key column "id" (using Parquet as an example) for resumable computation #

The following demonstrates how Daft's checkpoint feature automatically skips already processed records and processes only the remaining records when retrying failed Daft tasks.

• Applicable scenario: A dataset with a primary key (unique ID).
• Resumable computation requirement: When data processing fails, reruns should process only the remaining data to avoid duplication.
• Core configuration: Use checkpoint_config={"key_column": "id"}.

The complete code example is as follows.

from __future__ import annotations

import shutil
from pathlib import Path

import daft
from daft import col, set_runner_ray


def init_runner() -> None:
    try:
        set_runner_ray(noop_if_initialized=True)
        print("[info] Ray runner initialized")
    except Exception as e:
        print(f"[warn] Fallback to default runner: {e}")


def example_checkpoint_with_id() -> None:
    """
    Demonstrate Daft's checkpoint incremental write function:
    - Use the'id 'column as the unique key to avoid duplicate writes.
    - first write id=0..50(simulated historical data)
    - second write id=0..99(Simulate full pull)
    - Verify that the final result is 100 unique records.
    Note: The checkpoint function relies on the Ray executor.
    """
    # Construct test data
    df_all = daft.from_pydict({"id": list(range(100)), "val": [f"v{i}" for i in range(100)]})
    df_seed = df_all.where(col("id") <= 50)

    # Set the output path (clean before each run)
    dest = Path("/tmp/ckpt_demo_parquet_id")
    if dest.exists():
        shutil.rmtree(dest)
    dest.mkdir(parents=True, exist_ok=True)

    # Configure the checkpoint: use 'id' as the unique key
    ck = {"key_column": "id"}

    # First write: seed data (id <= 50)
    df_seed.write_parquet(str(dest), write_mode="append", checkpoint_config=ck)

    # Second write: full data (id 0~99), checkpoint automatically deduplicates
    df_all.write_parquet(str(dest), write_mode="append", checkpoint_config=ck)

    # Validate the result
    out_df = daft.read_parquet(str(dest))
    ids = out_df.select("id").to_pydict()["id"]
    print("[result] rows:", len(ids), "unique_ids:", len(set(ids)))
    print("[result] has_all_ids_0_to_99:", set(ids) == set(range(100)))


if __name__ == "__main__":
    init_runner()
    example_checkpoint_with_id()

Demo 2: Incremental computation based on file path #

The following demonstrates how Daft's checkpoint feature avoids redundant processing in incremental computation scenarios by processing only new files and skipping files that have already been processed.

• Applicable scenario: A continuously growing file directory (such as logs, images, or audio), where only newly added files are processed each time.
• Incremental requirement: Avoid processing files that have already been processed.
• Core configuration: Use checkpoint_config={"key_column": "path"}, using the file path as the unique identifier.

The complete code example is as follows.

from __future__ import annotations

import shutil
from pathlib import Path

import daft
from daft import set_runner_ray


def init_runner() -> None:
    try:
        set_runner_ray(noop_if_initialized=True)
        print("[info] Ray runner initialized")
    except Exception as e:
        print(f"[warn] Fallback to default runner: {e}")


def example_checkpoint_with_path() -> None:
    """
    Demonstrates incremental writing of checkpoints based on file paths:
    1. Create 10 initial files(file_0.txt ~ file_9.txt)
    2. Add 5 new files (file_10.txt ~ file_14.txt)；
    3. Rescan all files, automatically skip existing paths through checkpoints
    4. Validation ultimately consists of 15 unique paths with no duplicates.
    Note: The checkpoint function relies on the Ray executor.
    """
    src_dir = Path("/tmp/glob_src")
    dest = Path("/tmp/ckpt_demo_parquet_path")

    # Clean up old data to ensure repeatable runs
    if src_dir.exists():
        shutil.rmtree(src_dir)
    if dest.exists():
        shutil.rmtree(dest)
    src_dir.mkdir(parents=True)
    dest.mkdir(parents=True)

    # Step 1: Generate initial files (0-9)
    for i in range(10):
        (src_dir / f"file_{i}.txt").write_text("x" * i)

    df_initial = daft.from_glob_path(str(src_dir / "*.txt"))
    ck = {"key_column": "path"}
    df_initial.write_parquet(str(dest), write_mode="append", checkpoint_config=ck)

    # Step 2: Add new files (10-14)
    for i in range(10, 15):
        (src_dir / f"file_{i}.txt").write_text("x" * i)

    # Step 3: Rescan all files, checkpoint automatically deduplicates
    df_full = daft.from_glob_path(str(src_dir / "*.txt"))
    df_full.write_parquet(str(dest), write_mode="append", checkpoint_config=ck)

    # Step 4: Validate the result
    out_df = daft.read_parquet(str(dest))
    paths = out_df.select("path").to_pydict()["path"]
    print("[result] total rows:", len(paths), "unique paths:", len(set(paths)))

    expected_paths = {str(src_dir / f"file_{i}.txt") for i in range(15)}
    print("[result] has_all_expected_paths:", set(paths) == expected_paths)


if __name__ == "__main__":
    init_runner()
    example_checkpoint_with_path()