The business code and Python dependencies of Daft jobs change frequently. Packaging everything into the image requires frequent image updates, while placing the extracted dependencies directly on TOS results in cold starts being slowed down by a large number of small files. This article introduces the best practices solution: TOS only stores project.zip and Wheel archives. When connecting to Ray for the first time, Runtime Env delivers everything at once. The working_dir distributes the code, and pip creates a virtual environment on each Worker and caches it by hash. Jobs with the same configuration can use the cache on subsequent starts, improving startup efficiency.

Problems encountered #

When users run tasks using the unified Daft base image, there are two types of frequently changing business-side content:

• Additional Python Wheel dependencies.
• Business source code that needs to be imported by Daft UDF.

Common handling approaches lead to significant issues:

• Packaging everything into the image: Images are typically tens of GB in size. Every change to the business code requires rebuilding and pushing the image, resulting in high delivery costs and slow cold starts.
• Storing everything in TOS: Running with extracted Python dependencies placed directly in the TOS mount directory leads to a large number of small file reads, making cold starts even slower, and each task reloads everything. TOS is suitable for saving and transferring archives, but not as a runtime Python dependency directory.

Solution overview #

TOS only saves ZIP or Wheel artifacts. Distribution of code and dependencies is handled entirely by Ray Runtime Env and executed only once at the first Ray connection.
The division of roles is as follows:

• TOS: Saves project.zip and dependency Wheels, and does not upload extracted site-packages.
• Driver: Pulls the zip from TOS, copies it to the local disk, and extracts it.
• Ray Runtime Env: At the first Ray connection (during ray.init), delivers code and dependencies to each Worker in one go:
  • Code: Uses runtime_env.working_dir to point to the directory extracted by the Driver. Ray automatically synchronizes it to each Worker's local disk.
  • Dependencies: Uses runtime_env.pip to specify the Wheel list or requirements file. Ray runs pip install locally on each Worker to build an independent pip environment.
• Worker: Daft's Runner, Task, and Actor inherit the same default Runtime Env, importing code and dependencies from the local disk without accessing TOS during runtime.

Essential understanding: comparison between LAS run_envs and Ray runtime_env #

Data flow diagram #

Dependency and code layering #

Divide the assets used during job execution into four layers based on three dimensions: change frequency, size, and whether they are imported by Python, and place each in the most appropriate location.

project.zip
├── main.py
├── operators.py
└── requirements.txt

humanize==4.13.0

humanize  # 未指定版本
humanize>=4.13.0   # 允许安装更高版本

Why is this solution faster #

TOS only stores ZIP or Wheel files, and the distribution of code and dependencies is handled by Ray Runtime Env.

• TOS only transfers a single ZIP, and does not create, delete, or traverse thousands of small site-packages files.
• The Driver only extracts the business code on the local disk, so all small file I/O occurs locally on the node.
• working_dir is distributed once and cache is reused: Ray distributes and extracts working_dir into the local Runtime Env cache on each node, and multiple tasks within the same connection session reuse it.
• pip environment is isolated locally on each node: Dynamic Wheel files are installed into an independent pip environment on each node, cached by configuration hash, and the same configuration will hit the cache on the next startup.
• Daft's Runner/Task/Actor inherit the same default Runtime Env, so they can directly import business code and dynamic dependencies without repeated configuration.

Local node paths are similar to:

/tmp/ray/session.../runtime_resources/
├── pip/<hash>/virtualenv/lib/python3.10/site-packages/
└── working_dir_files/<package-hash>/

• There is still a cold start cost when Runtime Env is created for the first time, but the writes are to the node's local disk and can be cached by configuration hash, so there will not be a small file storm on TOS.
• working_dir is only for code and small configuration files, and should not contain datasets, model weights, or virtual environments.
• The local Runtime Env cache usually follows the lifecycle of the Ray node or Pod, and a cold start will still occur when a new cluster is created.
• The pip.conf in the base image, Mirror network, and private package authentication must take effect for all Workers.

Job examples #

requirements.txt

# Dynamic dependency list, handed to runtime_env.pip for installation on each Worker
humanize==4.13.0

operators.py

import json
import os
import time


def enrich_record(value: int) -> str:
    # Daft UDF: executed remotely on each Worker, returns a JSON record
    # Result fields used for validation: dependencies installed, code from local cache, tasks span multiple Workers
    # Place import inside function body: parse after runtime_env takes effect
    import humanize
    import operators
    import ray

    # Ensure partition execution time is long enough to observe multi-Worker concurrency
    time.sleep(0.25)
    return json.dumps({
        "humanized": humanize.intcomma(value * 1000),
        "humanize_file": humanize.__file__,          # Should reside in the node's local virtual environment
        "humanize_version": humanize.__version__,
        "node_id": ray.get_runtime_context().get_node_id(),
        "operators_file": operators.__file__,        # Should reside in the local cache of working_dir
        "runtime_env_marker": os.environ.get("LAS_NODE_LOCAL_RUNTIME_ENV"),
        "value": value,
    }, sort_keys=True)

main.py

import json
import os
from collections import Counter

import daft
import ray

from operators import enrich_record


# Driver entry command has already extracted project.zip to this directory
PROJECT_DIR = os.environ["LOCAL_PROJECT_DIR"]

# Parse requirements.txt to obtain pip package list
with open(os.path.join(PROJECT_DIR, "requirements.txt")) as file:
    packages = [
        line.strip()
        for line in file
        if line.strip() and not line.lstrip().startswith("#")
    ]

# Core of the solution: working_dir distributes code; pip creates a local virtual environment and caches by hash on each Worker; env_vars are delivered accordingly
runtime_env = {
    "working_dir": PROJECT_DIR,
    "pip": {
        "packages": packages,
        # This image has historical conflicts, disable full environment check; when the image is clean, it is recommended to revert to True
        "pip_check": False,
    },
    "env_vars": {
        "LAS_NODE_LOCAL_RUNTIME_ENV": "enabled",
    },
}

# If the platform/SDK has already initialized Ray, runtime_env will not take effect and will fail directly
if ray.is_initialized():
    raise RuntimeError(
        "Ray was initialized before runtime_env was configured; "
        "configure runtime_env at the platform's first ray.init() instead."
    )

# Connect to the cluster already started by LAS, deliver runtime_env to all Workers at this moment in one go
ray.init(address="auto", runtime_env=runtime_env)
daft.set_runner_ray()

# 24 records / 24 partitions, run UDF to obtain the enriched column
result = (
    daft.from_pydict({"value": list(range(24))})
    .repartition(24)
    .with_column(
        "enriched",
        daft.col("value").apply(
            enrich_record,
            return_dtype=daft.DataType.string(),
        ),
    )
    .collect()
    .to_pydict()
)

records = [json.loads(item) for item in result["enriched"]]
node_counts = Counter(record["node_id"] for record in records)

# End-to-end assertion: data is complete, ≥3 Workers, dependencies/code/env_vars all come from runtime_env and no longer revisit TOS
assert len(records) == 24
assert len(node_counts) >= 3
assert all(record["humanize_version"] == "4.13.0" for record in records)
assert all(record["runtime_env_marker"] == "enabled" for record in records)
assert all("/intelligent-data/" not in record["humanize_file"] for record in records)
assert all("/intelligent-data/" not in record["operators_file"] for record in records)

print("worker_node_counts:", dict(node_counts))
print("sample:", records[0])
print("LAS_NODE_LOCAL_RUNTIME_ENV_E2E_SUCCESS")

Submission examples #

Submit via LAS API
Pre-run setup:

# Install Volcano Engine Python SDK (including LAS 2025-08-15 API)
python -m pip install volcengine-python-sdk

# LAS API and TOS mount share the same set of AK/SK; use key management in production
export ACCESS_KEY="<your_ak>"
export SECRET_KEY="<your_sk>"

# LAS resource pool and TOS archive prefix (only project.zip is placed under the prefix)
export LAS_RESOURCE_ID="<LAS 队列或独占资源 ID>"
export TOS_ARTIFACT_URI="tos://<your_bucket>/<path>/"

# CPU image used for verification in this document; account must be able to pull
export LAS_DAFT_IMAGE="las-ai-online-cn-beijing.cr.volces.com/las/ve-daft-pylance7:0.7.21.post1-py3.10-ubuntu24.04"

# First, use CreateTask to create a task template, then use RunTaskInstance to launch an instance
python submit_daft_job.py

The following submission script only mounts a single project.zip from TOS to the task. The entry command copies the ZIP to the Driver's local disk and extracts it; dependencies are not installed on TOS, and no small project files are written back to TOS.

import os
import uuid

from volcenginesdkcore import ApiClient, Configuration
from volcenginesdklas20250815 import LAS20250815Api
from volcenginesdklas20250815.models import (
    CreateTaskRequest,
    DaftResourceConfigForRunTaskInstanceInput,
    DaftTaskForCreateTaskInput,
    EnvForCreateTaskInput,
    RunEnvForRunTaskInstanceInput,
    RunTaskInstanceRequest,
    TosMountForCreateTaskInput,
    TosMountPathForCreateTaskInput,
)


# AK/SK are used for both LAS API authentication and TOS CSI mounting
ACCESS_KEY = os.environ["ACCESS_KEY"]
SECRET_KEY = os.environ["SECRET_KEY"]

# Resource ID, image, TOS archive prefix
RESOURCE_ID = os.environ["LAS_RESOURCE_ID"]
IMAGE = os.environ["LAS_DAFT_IMAGE"]
TOS_SOURCE = os.environ["TOS_ARTIFACT_URI"]

REGION = "cn-beijing"
LAS_HOST = "open.volcengineapi.com"

# ARTIFACT_ROOT: TOS read-only mount point; LOCAL_ROOT: Driver local disk, used for decompression and execution
ARTIFACT_ROOT = "/intelligent-data/artifacts/my_daft_job"
LOCAL_ROOT = "/tmp/my_daft_job"
LOCAL_PROJECT_DIR = f"{LOCAL_ROOT}/project"

# Initialize LAS OpenAPI client
config = Configuration()
config.ak = ACCESS_KEY
config.sk = SECRET_KEY
config.region = REGION
config.host = LAS_HOST
config.scheme = "https"
api = LAS20250815Api(ApiClient(config))

# Driver entry command: copy zip to local → decompress locally → run main.py
# Insert three timestamps in the process to facilitate splitting cold start time later
entrypoint = (
    "bash -lc 'set -euo pipefail; "
    "date \"+ARTIFACT_COPY_START=%F %T\"; "
    "mkdir -p \"$LOCAL_PROJECT_DIR\"; "
    "cp \"$ARTIFACT_ROOT/project.zip\" \"$LOCAL_ROOT/project.zip\"; "
    "date \"+ARTIFACT_COPY_DONE=%F %T\"; "
    "unzip -o -q \"$LOCAL_ROOT/project.zip\" -d \"$LOCAL_PROJECT_DIR\"; "
    "date \"+LOCAL_UNZIP_DONE=%F %T\"; "
    "python \"$LOCAL_PROJECT_DIR/main.py\"'"
)

# Step 1: Use CreateTask to define a reusable Daft task template and mount the TOS prefix to ARTIFACT_ROOT
created = api.create_task(
    CreateTaskRequest(
        client_token=str(uuid.uuid4()),  # Idempotent token
        name="daft-node-local-runtime-env",
        description="Distribute code and wheels to Ray node-local environments",
        task_type="Daft",
        timeout_minutes=30,
        daft_task=DaftTaskForCreateTaskInput(
            image=IMAGE,
            image_type="BuiltIn",
            entrypoint=entrypoint,
            envs=[EnvForCreateTaskInput(key="PYTHONUNBUFFERED", value="1")],
        ),
        tos_mount=TosMountForCreateTaskInput(
            mount_type="Csi",
            access_key=ACCESS_KEY,
            secret_key=SECRET_KEY,
            tos_mount_paths=[
                TosMountPathForCreateTaskInput(
                    source=TOS_SOURCE,
                    mount_path=f"{ARTIFACT_ROOT}/",
                )
            ],
        ),
    )
)

# Step 2: Use RunTaskInstance to launch an instance
# run_envs only passes path variables; actual pip/working_dir is set by ray.init(runtime_env=...) in main.py
# 1 Head + 3 Workers, each node 1C4G; for GPU scenarios, change worker_gpu_num and use a GPU image
started = api.run_task_instance(
    RunTaskInstanceRequest(
        client_token=str(uuid.uuid4()),
        task_id=created.task_id,
        resource_id=RESOURCE_ID,
        resource_spec_group_id="0",
        run_envs=[
            RunEnvForRunTaskInstanceInput(key="ARTIFACT_ROOT", value=ARTIFACT_ROOT),
            RunEnvForRunTaskInstanceInput(key="LOCAL_ROOT", value=LOCAL_ROOT),
            RunEnvForRunTaskInstanceInput(key="LOCAL_PROJECT_DIR", value=LOCAL_PROJECT_DIR),
        ],
        daft_resource_config=DaftResourceConfigForRunTaskInstanceInput(
            head_cpu_num=1,
            head_memory_size=4,
            worker_cpu_num=1,
            worker_memory_size=4,
            worker_gpu_num=0,
            replicas=3,
        ),
    )
)

print("task_id:", created.task_id)
print("task_instance_id:", started.task_instance_id)

• Runtime Env creation failed: Check Mirror connectivity, whether the Wheel supports the current Python/CPU architecture, and whether there are unrelated pip check conflicts in the base image.
• Driver runs but Worker import fails: Ensure the runtime_env configuration takes effect at the first ray.init(), and do not only modify the Driver's sys.path.
• Business code not updated: Change the version path or content hash of project.zip to avoid reusing old cache.
• Dependencies are still taking a long time to install: Move stable, large dependencies into the base image, and ensure all Workers can access the customer Mirror; do not revert to the TOS --target directory.