deepdata/runtime/python-executor/README-zh.md

Ray 执行器

概述

Ray 执行器是基于 Ray 的分布式执行框架，负责执行数据处理算子、任务调度和分布式计算。

架构

runtime/python-executor/
└── deepdata/
    ├── core/
    │   ├── base_op.py      # BaseOp, Mapper, Filter, Slicer, LLM
    │   ├── dataset.py      # Dataset 处理
    │   └── constant.py     # 常量定义
    ├── scheduler/
    │   ├── scheduler.py    # TaskScheduler, Task, TaskStatus
    │   ├── func_task_scheduler.py   # 函数任务调度
    │   └── cmd_task_scheduler.py    # 命令任务调度
    ├── wrappers/
    │   ├── executor.py     # Ray 执行器入口
    │   ├── deepdata_wrapper.py      # DeepData 任务包装
    │   └── data_juicer_wrapper.py   # DataJuicer 集成
    └── common/utils/       # 工具函数
        ├── bytes_transform.py
        ├── file_scanner.py
        ├── lazy_loader.py
        └── text_splitter.py

组件

1. Base 类

BaseOp

所有算子的基类：

class BaseOp:
    def __init__(self, *args, **kwargs):
        self.accelerator = kwargs.get('accelerator', "cpu")
        self.text_key = kwargs.get('text_key', "text")
        # ... 其他配置
    
    def execute(self, sample):
        raise NotImplementedError

Mapper

数据转换算子基类（1:1）：

class Mapper(BaseOp):
    def execute(self, sample: Dict) -> Dict:
        # 转换逻辑
        return processed_sample

Filter

数据过滤算子基类（返回 bool）：

class Filter(BaseOp):
    def execute(self, sample: Dict) -> bool:
        # 过滤逻辑
        return True  # 保留或过滤

Slicer

数据切片算子基类（1:N）：

class Slicer(BaseOp):
    def execute(self, sample: Dict) -> List[Dict]:
        # 切片逻辑
        return [sample1, sample2, ...]

LLM

LLM 算子基类：

class LLM(Mapper):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.llm = self.get_llm(*args, **kwargs)
    
    def build_llm_prompt(self, *args, **kwargs):
        raise NotImplementedError

2. Task Scheduler

异步任务调度器：

class TaskScheduler:
    def __init__(self, max_concurrent: int = 10):
        self.tasks: Dict[str, Task] = {}
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def submit(self, task_id, task, *args, **kwargs):
        # 提交任务
        pass
    
    def get_task_status(self, task_id: str) -> Optional[TaskResult]:
        # 获取任务状态
        pass
    
    def cancel_task(self, task_id: str) -> bool:
        # 取消任务
        pass

3. 算子执行

算子注册

from deepdata.core.base_op import OPERATORS

OPERATORS.register_module(
    module_name='YourOperatorName',
    module_path="ops.user.operator_package.process"
)

执行算子

from deepdata.core.base_op import Mapper

class MyMapper(Mapper):
    def execute(self, sample):
        text = sample.get('text', '')
        processed = text.upper()
        sample['text'] = processed
        return sample

快速开始

前置条件

• Python 3.11+
• Ray 2.7.0+
• Poetry

安装

cd runtime/python-executor
poetry install

启动 Ray Head

ray start --head

启动 Ray Worker

ray start --head-address=<head-ip>:6379

使用

提交任务到 Ray

from ray import remote

@remote
def execute_operator(sample, operator_config):
    # 执行算子逻辑
    return result

# 提交任务
result_ref = execute_operator.remote(sample, config)
result = ray.get(result_ref)

使用 Task Scheduler

from deepdata.scheduler.scheduler import TaskScheduler

scheduler = TaskScheduler(max_concurrent=10)
task_id = "task-001"
scheduler.submit(task_id, my_function, arg1, arg2)
status = scheduler.get_task_status(task_id)

开发

添加新算子

1. 在 runtime/ops/ 创建算子目录
2. 实现 process.py 和 __init__.py
3. 在 __init__.py 注册算子
4. 测试算子

调试算子

# 本地测试
python -c "from ops.user.operator_package.process import YourOperatorName; op = YourOperatorName(); print(op.execute({'text': 'test'}))"

性能

并行执行

Ray 自动处理并行执行和资源分配。

容错

Ray 提供自动任务重试和故障转移。

资源管理

Ray 动态分配 CPU、GPU、内存资源。

文档

• Ray 文档
• AGENTS.md

相关链接

• 运行时 README
• 算子生态