Daft 作业的业务代码与 Python 依赖时常变动，若全量打包进镜像，会需要频繁修改镜像，若把解压后的依赖直接放 TOS 上，冷启动会被大量小文件拖慢。本文为您介绍最佳实践方案：TOS 只存 project.zip 和 Wheel 归档，首次连接 Ray 时利用 Runtime Env 一次性下发，working_dir 分发代码，pip 在每个 Worker 本地建虚拟环境并按哈希缓存，同配置的作业再次启动即可使用缓存，提高启动效率。

遇到的问题 #

用户使用统一的 Daft 基础镜像运行任务时，业务侧有两类内容变化频繁：

• 额外的 Python Wheel 依赖。
• 需要被 Daft UDF import 的业务源码。

通常的处理方案都会引发明显问题：

• 全打包进镜像：镜像通常十几 GB，业务代码每次修改都要重新构建镜像并推送，交付成本高、冷启动慢。
• 全存进 TOS：把展开后的 Python 依赖直接放在 TOS 挂载目录里运行，会产生大量小文件读取，冷启动更慢，且每次任务都要重新加载。TOS 适合保存和传输归档，不适合作为运行时的 Python 依赖目录。

方案概述 #

TOS 只保存 ZIP 或 Wheel 制品；代码与依赖的分发全部交给 Ray Runtime Env，且只在首次连接 Ray 时执行一次。
各角色分工如下：

• TOS： 保存 project.zip 和依赖 Wheel，不上传解压后的 site-packages。
• Driver： 从 TOS 拉取 zip，复制到本地盘并解压。
• Ray Runtime Env： 在首次连接 Ray 时（ray.init 时刻），一次性把代码和依赖下发到每一个 Worker：
  • 代码： 通过 runtime_env.working_dir 指向 Driver 解压后的目录，Ray 自动同步到每个 Worker 本地磁盘。
  • 依赖： 通过 runtime_env.pip 指定 Wheel 列表或 requirements 文件，Ray 在每个 Worker 本地执行 pip install，构建独立的 pip 环境。
• Worker： Daft 的 Runner、Task、Actor 继承同一个默认 Runtime Env，从本地磁盘 import 代码和依赖，运行期不再访问 TOS。

必备理解：LAS run_envs 与 Ray runtime_env 对比 #

数据流示意 #

依赖和代码分层 #

按变化频率、体积、是否被 Python import 三个维度，把作业运行时用到的资产分成 4 层，各自放到最合适的位置。

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

humanize==4.13.0

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

为什么这个方案更快 #

TOS 只放 ZIP 或 Wheel，代码与依赖的分发交给 Ray Runtime Env。

• TOS 只传输一个 ZIP，不会创建、删除或遍历成千上万个 site-packages 小文件。
• Driver 只在本地盘解压业务代码，所有小文件 I/O 都发生在节点本地。
• working_dir 分发一次，缓存复用： Ray 把 working_dir 分发并解压到每个节点的本地 Runtime Env 缓存，同一连接期内多次任务复用。
• pip 环境节点本地隔离： 动态 Wheel 安装到每个节点本地的独立 pip 环境，按配置哈希缓存，同配置下次启动即命中。
• Daft 的 Runner/Task/Actor 继承同一个默认 Runtime Env，因此能够直接 import 业务代码和动态依赖，无需重复配置。

节点本地路径类似于：

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

• Runtime Env 首次创建仍有冷启动成本，但写入的是节点本地盘，并且可以按配置哈希缓存，不会在 TOS 上制造小文件风暴。
• working_dir 只用于代码和小型配置，不应包含数据集、模型权重或虚拟环境。
• 本地 Runtime Env 缓存通常跟随 Ray 节点或 Pod 生命周期，新建集群仍会发生一次冷启动。
• 基础镜像中的 pip.conf、Mirror 网络和私有包认证必须对所有 Worker 生效。

作业示例 #

requirements.txt

# 动态依赖清单，交给 runtime_env.pip 装到每个 Worker
humanize==4.13.0

operators.py

import json
import os
import time


def enrich_record(value: int) -> str:
    # Daft UDF：每个 Worker 上远程执行，返回一条 JSON
    # 结果字段用于验证：依赖已装、代码来自本地缓存、任务跨多 Worker
    # import 放函数体内：等 runtime_env 生效后再解析
    import humanize
    import operators
    import ray

    # 让分区执行时间足够长，便于观察多 Worker 并发
    time.sleep(0.25)
    return json.dumps({
        "humanized": humanize.intcomma(value * 1000),
        "humanize_file": humanize.__file__,          # 应位于节点本地虚拟环境
        "humanize_version": humanize.__version__,
        "node_id": ray.get_runtime_context().get_node_id(),
        "operators_file": operators.__file__,        # 应位于 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 入口命令已把 project.zip 解压到该目录
PROJECT_DIR = os.environ["LOCAL_PROJECT_DIR"]

# 解析 requirements.txt，得到 pip 包列表
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("#")
    ]

# 方案核心：working_dir 分发代码；pip 在每个 Worker 本地建虚拟环境并按哈希缓存；env_vars 随之下发
runtime_env = {
    "working_dir": PROJECT_DIR,
    "pip": {
        "packages": packages,
        # 本次镜像有历史冲突，关闭全环境检查；镜像干净时建议改回 True
        "pip_check": False,
    },
    "env_vars": {
        "LAS_NODE_LOCAL_RUNTIME_ENV": "enabled",
    },
}

# 若平台/SDK 已经先 init 过 Ray，runtime_env 不会生效，直接失败
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."
    )

# 连接 LAS 已拉起的集群，此刻一次性把 runtime_env 下发到所有 Worker
ray.init(address="auto", runtime_env=runtime_env)
daft.set_runner_ray()

# 24 条数据 / 24 分区，跑 UDF 得到 enriched 列
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)

# 端到端断言：数据完整、≥3 Worker、依赖/代码/env_vars 均来自 runtime_env 且不再回访 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")

提交示例 #

通过 LAS API 提交
运行前设置：

# 安装火山引擎 Python SDK（含 LAS 2025-08-15 API）
python -m pip install volcengine-python-sdk

# LAS API 与 TOS 挂载共用同一组 AK/SK；生产请走密钥托管
export ACCESS_KEY="<your_ak>"
export SECRET_KEY="<your_sk>"

# LAS 资源池与 TOS 归档前缀（前缀下只放 project.zip）
export LAS_RESOURCE_ID="<LAS 队列或独占资源 ID>"
export TOS_ARTIFACT_URI="tos://<your_bucket>/<path>/"

# 本文验证使用的 CPU 镜像；账号需可拉取
export LAS_DAFT_IMAGE="las-ai-online-cn-beijing.cr.volces.com/las/ve-daft-pylance7:0.7.21.post1-py3.10-ubuntu24.04"

# 先 CreateTask 建任务模板，再 RunTaskInstance 拉起实例
python submit_daft_job.py

下面的提交脚本只把一个 project.zip 从 TOS 挂载到任务中。入口命令将 ZIP 复制到 Driver 本地盘并解压，不在 TOS 上安装依赖，也不向 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 同时用于 LAS API 认证和 TOS CSI 挂载
ACCESS_KEY = os.environ["ACCESS_KEY"]
SECRET_KEY = os.environ["SECRET_KEY"]

# 资源 ID、镜像、TOS 归档前缀
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 只读挂载点；LOCAL_ROOT：Driver 本地盘，用于解压和运行
ARTIFACT_ROOT = "/intelligent-data/artifacts/my_daft_job"
LOCAL_ROOT = "/tmp/my_daft_job"
LOCAL_PROJECT_DIR = f"{LOCAL_ROOT}/project"

# 初始化 LAS OpenAPI 客户端
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 入口命令：复制 zip 到本地 → 本地解压 → 运行 main.py
# 中间打三个时间戳，便于事后拆分冷启动耗时
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\"'"
)

# 第一步：CreateTask 定义可复用的 Daft 任务模板，把 TOS 前缀挂到 ARTIFACT_ROOT
created = api.create_task(
    CreateTaskRequest(
        client_token=str(uuid.uuid4()),  # 幂等 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}/",
                )
            ],
        ),
    )
)

# 第二步：RunTaskInstance 拉起实例
# run_envs 只传路径变量；真正的 pip/working_dir 由 main.py 里的 ray.init(runtime_env=...) 完成
# 1 Head + 3 Worker，每节点 1C4G；GPU 场景改 worker_gpu_num 并换 GPU 镜像
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 创建失败：​检查 Mirror 连通性、Wheel 是否支持当前 Python/CPU 架构，以及基础镜像是否存在无关的 pip check 冲突。
• Driver 能运行、Worker import 失败：​确认 runtime_env 配置在第一次 ray.init() 时生效，不要只修改 Driver 的 sys.path。
• 业务代码没有更新：​更换 project.zip 的版本路径或内容哈希，避免复用旧缓存。
• 依赖仍然安装很久：​把稳定大依赖移入基础镜像，并确认所有 Worker 都能访问客户 Mirror；不要回退到 TOS --target 目录。