deepdata: core product code (backend, frontend, runtime, deepflow module)

This commit is contained in:
figmar
2026-08-30 19:23:38 +08:00
commit bd3db30be9
1405 changed files with 271727 additions and 0 deletions
+221
View File
@@ -0,0 +1,221 @@
# 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
所有算子的基类:
```python
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):
```python
class Mapper(BaseOp):
def execute(self, sample: Dict) -> Dict:
# 转换逻辑
return processed_sample
```
#### Filter
数据过滤算子基类(返回 bool):
```python
class Filter(BaseOp):
def execute(self, sample: Dict) -> bool:
# 过滤逻辑
return True # 保留或过滤
```
#### Slicer
数据切片算子基类(1:N):
```python
class Slicer(BaseOp):
def execute(self, sample: Dict) -> List[Dict]:
# 切片逻辑
return [sample1, sample2, ...]
```
#### LLM
LLM 算子基类:
```python
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
异步任务调度器:
```python
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. 算子执行
#### 算子注册
```python
from deepdata.core.base_op import OPERATORS
OPERATORS.register_module(
module_name='YourOperatorName',
module_path="ops.user.operator_package.process"
)
```
#### 执行算子
```python
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
### 安装
```bash
cd runtime/python-executor
poetry install
```
### 启动 Ray Head
```bash
ray start --head
```
### 启动 Ray Worker
```bash
ray start --head-address=<head-ip>:6379
```
## 使用
### 提交任务到 Ray
```python
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
```python
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. 测试算子
### 调试算子
```bash
# 本地测试
python -c "from ops.user.operator_package.process import YourOperatorName; op = YourOperatorName(); print(op.execute({'text': 'test'}))"
```
## 性能
### 并行执行
Ray 自动处理并行执行和资源分配。
### 容错
Ray 提供自动任务重试和故障转移。
### 资源管理
Ray 动态分配 CPU、GPU、内存资源。
## 文档
- [Ray 文档](https://docs.ray.io/)
- [AGENTS.md](./AGENTS.md)
## 相关链接
- [运行时 README](../README.md)
- [算子生态](../ops/README.md)
+221
View File
@@ -0,0 +1,221 @@
# Ray Executor
## Overview
Ray Executor is a Ray-based distributed execution framework responsible for executing data processing operators, task scheduling, and distributed computing.
## Architecture
```
runtime/python-executor/
└── deepdata/
├── core/
│ ├── base_op.py # BaseOp, Mapper, Filter, Slicer, LLM
│ ├── dataset.py # Dataset processing
│ └── constant.py # Constant definitions
├── scheduler/
│ ├── scheduler.py # TaskScheduler, Task, TaskStatus
│ ├── func_task_scheduler.py # Function task scheduling
│ └── cmd_task_scheduler.py # Command task scheduling
├── wrappers/
│ ├── executor.py # Ray executor entry point
│ ├── deepdata_wrapper.py # DeepData task wrapper
│ └── data_juicer_wrapper.py # DataJuicer integration
└── common/utils/ # Utility functions
├── bytes_transform.py
├── file_scanner.py
├── lazy_loader.py
└── text_splitter.py
```
## Components
### 1. Base Classes
#### BaseOp
Base class for all operators:
```python
class BaseOp:
def __init__(self, *args, **kwargs):
self.accelerator = kwargs.get('accelerator', "cpu")
self.text_key = kwargs.get('text_key', "text")
# ... other configuration
def execute(self, sample):
raise NotImplementedError
```
#### Mapper
Base class for data transformation operators (1:1):
```python
class Mapper(BaseOp):
def execute(self, sample: Dict) -> Dict:
# Transformation logic
return processed_sample
```
#### Filter
Base class for data filtering operators (returns bool):
```python
class Filter(BaseOp):
def execute(self, sample: Dict) -> bool:
# Filtering logic
return True # Keep or filter out
```
#### Slicer
Base class for data slicing operators (1:N):
```python
class Slicer(BaseOp):
def execute(self, sample: Dict) -> List[Dict]:
# Slicing logic
return [sample1, sample2, ...]
```
#### LLM
Base class for LLM operators:
```python
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
Async task scheduler:
```python
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):
# Submit task
pass
def get_task_status(self, task_id: str) -> Optional[TaskResult]:
# Get task status
pass
def cancel_task(self, task_id: str) -> bool:
# Cancel task
pass
```
### 3. Operator Execution
#### Operator Registration
```python
from deepdata.core.base_op import OPERATORS
OPERATORS.register_module(
module_name='YourOperatorName',
module_path="ops.user.operator_package.process"
)
```
#### Execute Operator
```python
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
```
## Quick Start
### Prerequisites
- Python 3.11+
- Ray 2.7.0+
- Poetry
### Installation
```bash
cd runtime/python-executor
poetry install
```
### Start Ray Head
```bash
ray start --head
```
### Start Ray Worker
```bash
ray start --head-address=<head-ip>:6379
```
## Usage
### Submit Task to Ray
```python
from ray import remote
@remote
def execute_operator(sample, operator_config):
# Execute operator logic
return result
# Submit task
result_ref = execute_operator.remote(sample, config)
result = ray.get(result_ref)
```
### Use Task Scheduler
```python
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)
```
## Development
### Adding a New Operator
1. Create operator directory in `runtime/ops/`
2. Implement `process.py` and `__init__.py`
3. Register operator in `__init__.py`
4. Test the operator
### Debugging Operators
```bash
# Local test
python -c "from ops.user.operator_package.process import YourOperatorName; op = YourOperatorName(); print(op.execute({'text': 'test'}))"
```
## Performance
### Parallel Execution
Ray automatically handles parallel execution and resource allocation.
### Fault Tolerance
Ray provides automatic task retry and failover.
### Resource Management
Ray dynamically allocates CPU, GPU, and memory resources.
## Documentation
- [Ray Documentation](https://docs.ray.io/)
- [AGENTS.md](./AGENTS.md)
## Related Links
- [Runtime README](../README.md)
- [Operator Ecosystem](../ops/README.md)
@@ -0,0 +1 @@
__version__ = "0.0.1"
@@ -0,0 +1,99 @@
# -*- coding: utf-8 -*-
"""
The class hierarchy for built-in exceptions is:
see https://docs.python.org/3/library/exceptions.html
BaseException
├── SystemExit
├── KeyboardInterrupt
├── GeneratorExit
└── Exception
├── ArithmeticError
│ ├── FloatingPointError
│ ├── OverflowError
│ └── ZeroDivisionError
├── AssertionError
├── AttributeError
├── BufferError
├── EOFError
├── ImportError
│ └── ModuleNotFoundError
├── LookupError
│ ├── IndexError
│ └── KeyError
├── MemoryError
├── NameError
│ └── UnboundLocalError
├── OSError
│ ├── BlockingIOError
│ ├── ChildProcessError
│ ├── ConnectionError
│ │ ├── BrokenPipeError
│ │ ├── ConnectionAbortedError
│ │ ├── ConnectionRefusedError
│ │ └── ConnectionResetError
│ ├── FileExistsError
│ ├── FileNotFoundError
│ ├── InterruptedError
│ ├── IsADirectoryError
│ ├── NotADirectoryError
│ ├── PermissionError
│ ├── ProcessLookupError
│ └── TimeoutError
├── ReferenceError
├── RuntimeError
│ ├── NotImplementedError
│ └── RecursionError
├── SyntaxError
│ └── IndentationError
│ └── TabError
├── SystemError
├── TypeError
├── ValueError
│ └── UnicodeError
│ ├── DecodeError
│ ├── EncodeError
│ └── UnicodeTranslateError
└── Warning
├── DeprecationWarning
├── PendingDeprecationWarning
├── RuntimeWarning
├── SyntaxWarning
├── UserWarning
├── FutureWarning
├── ImportWarning
└── UnicodeWarning
"""
from enum import Enum
ERROR_CODE_TABLE = {
ImportError: "ops.0001",
ModuleNotFoundError: "ops.0002",
NameError: "ops.0003",
KeyError: "ops.0004",
IndexError: "ops.0005",
ValueError: "ops.0006",
TypeError: "ops.0007",
SyntaxError: "ops.0008",
AttributeError: "ops.0009",
ArithmeticError: "ops.0010",
MemoryError: "ops.0011",
OSError: "ops.0012",
FileNotFoundError: "ops.0013",
NotADirectoryError: "ops.0014",
PermissionError: "ops.0015",
TimeoutError: "ops.0016",
}
UNKNOWN_ERROR_CODE = "ops.9999"
class ErrorCode(Enum):
# 通用错误
SUCCESS = (0, "Success")
UNKNOWN_ERROR = (1, "Unknown error")
PARAM_ERROR = (2, "Invalid parameter")
FILE_NOT_FOUND_ERROR = (1000, "File not found!")
SUBMIT_TASK_ERROR = (1001, "Task submitted Failed!")
CANCEL_TASK_ERROR = (1002, "Task canceled Failed!")
@@ -0,0 +1,63 @@
# -*- coding: utf-8 -*-
from datetime import datetime
import cv2
import numpy as np
import os
import pytz
from loguru import logger
def check_valid_path(file_path):
full_path = os.path.abspath(file_path)
return os.path.exists(full_path)
def get_realpath_with_prefix_check(path, prefix):
realpath = os.path.realpath(path)
if realpath.startswith(prefix):
return realpath
else:
raise ValueError(f"The path {realpath} does not start with the prefix '{prefix}'.")
def bytes_to_numpy(image_bytes):
"""bytes转数组"""
image_np = np.frombuffer(image_bytes, dtype=np.uint8)
image_np2 = cv2.imdecode(image_np, cv2.IMREAD_COLOR)
return image_np2
def numpy_to_bytes(image_np, file_type):
"""数组转bytes"""
if not image_np.size:
return b""
data = cv2.imencode(file_type, image_np)[1]
image_bytes = data.tobytes()
return image_bytes
def get_now_time(timezone, time_format, file_name, method):
timestamp = ""
try:
china_tz = pytz.timezone(timezone) # 设置时区
china_time = datetime.now(tz=china_tz) # 获取当前时间并转换为对应的时区
timestamp = china_time.strftime(time_format) # 格式化输出时间
except ValueError as e:
logger.error("fileName: %s, method: %s, formatting time failed: %s", file_name, method, e, exc_info=True)
return timestamp
def decrypt(enc_pass):
import kmc.kmc as K
os.environ['KMC_DATA_USER'] = 'modelenginepublic'
os.environ['KMC_PYTHON_ENCRYPT_DATA'] = enc_pass
dec_pass = K.API().decrypt(0)
os.environ['KMC_PYTHON_ENCRYPT_DATA'] = ""
return dec_pass
def is_k8s():
return "KUBERNETES_SERVICE_HOST" in os.environ
@@ -0,0 +1,115 @@
# -- encoding: utf-8 --
from collections import deque
class TrieNode:
def __init__(self, value):
self.value = value
self.child = dict()
self.fail = None
self.word = None
class AhoCorasic:
"""AC自动机算法进行目标字符串搜索"""
def __init__(self, words):
self._root = add_fail_pointer(build_trie(words))
def search(self, text: str, special_symbols: set):
"""
匹配敏感词。
Args:
text: 文本
special_symbols: 特殊字符(需跳过)
Returns:
匹配成功的字符串列表
"""
seq_list = []
node = self._root
valid_len = 0 # 当前遍历的有效长度
for i, s in enumerate(text):
if s in special_symbols: # 跳过特殊字符
if valid_len != 0:
valid_len += 1
continue
matched = True
while s not in node.child: # 当node.child没有字符s
if node == self._root: # 当node为root(无node.fail),有效长度归0且跳出
valid_len = 0
matched = False
break
elif node.fail == self._root: # node.fail为root场景,有效长度归0,但可继续
valid_len = 0
node = node.fail # 移动到失败指针节点
if not matched:
continue
node = node.child.get(s)
valid_len += 1
if node.word: # node是单词尾字母
sensitive_word = text[i - valid_len + 1:i + 1]
seq_list.append(sensitive_word)
seq_list = list(set(seq_list))
return seq_list
def build_trie(words: list):
"""
构建前缀树。
Args:
words: 敏感词列表。
Returns:
前缀树根节点。
"""
root = TrieNode('root')
for word in words:
node = root
for s in word:
if s not in node.child:
node.child[s] = TrieNode(s)
node = node.child[s]
if not node.word:
node.word = {word}
else:
node.word.add(word)
return root
def add_fail_pointer(root: TrieNode):
"""
为前缀树添加失败指针。
步骤:
1. 从root开始逐层将node和node.parent以二元组存放队列。root没有fail指针,root.child的失败指针即为root。
2. 对于root和root.child以外的node,查询node.parent.fail.child。
3. 如果存在node.parent.fail.child.value == node.value,则构建node.fail = node.parent.fail.child.value。
Args:
root: 前缀树根节点。
returns:
添加失败指针后的前缀树根节点。
"""
queue = deque()
queue.appendleft((root, None))
while len(queue) > 0:
node_parent = queue.pop()
curr, parent = node_parent[0], node_parent[1]
for sub in curr.child.values():
queue.appendleft((sub, curr))
if parent is None:
continue
elif parent is root:
curr.fail = root
else:
parent_fail = parent.fail
while parent_fail and curr.value not in parent_fail.child:
parent_fail = parent_fail.fail
if parent_fail:
curr.fail = parent_fail.child[curr.value]
else:
curr.fail = root
return root
@@ -0,0 +1,66 @@
# -- encoding: utf-8 --
import pickle
import base64
from io import BytesIO
import cv2
import numpy as np
from PIL import Image
def bytes_to_numpy(image_bytes):
"""bytes转数组"""
image_np = np.frombuffer(image_bytes, dtype=np.uint8)
image_np2 = cv2.imdecode(image_np, cv2.IMREAD_COLOR)
return image_np2
def numpy_to_bytes(image_np, file_type):
"""
数组转bytes
Params:
file_type: as required by OpenCV, extension must have a leading period.
"""
if not image_np.size:
return b""
data = cv2.imencode(file_type, image_np)[1]
image_bytes = data.tobytes()
return image_bytes
def pil_to_bytes(src: Image.Image) -> bytes:
"""将 PIL.Image 转换为字节流"""
# 确保图像是 RGB 模式
src = src.convert("RGB")
with BytesIO() as bytes_io:
src.save(bytes_io, format='PNG')
im_bytes = bytes_io.getvalue()
return im_bytes
def bytes_to_pil(src: bytes) -> Image.Image:
"""将字节流转换为 PIL.Image"""
with BytesIO() as bytes_io:
with Image.open(bytes_io) as pil_img: # 使用with/as语句确保资源被正确释放
pil_img.load() # 确保图像数据被加载
return pil_img.copy() # 返回图像的副本以避免资源被关闭后无法使用
def pil_to_base64(src: Image.Image):
"""PIl.Image转base64"""
with BytesIO() as img_buffer:
src.save(img_buffer, format='png')
byte_data = img_buffer.getvalue()
base64_str = base64.b64encode(byte_data)
return base64_str
def obj_to_bytes(src: object) -> bytes:
return pickle.dumps(src)
def bytes_to_obj(src: bytes) -> object:
return pickle.loads(src)
@@ -0,0 +1,30 @@
# -*- coding: utf-8 -*-
import importlib.abc
import importlib.util
from pathlib import Path
class CustomImporter(importlib.abc.MetaPathFinder):
def __init__(self, base_path):
self.base_path = Path(base_path).resolve()
def find_spec(self, fullname, path, target=None):
# 将模块名转换为路径(例如:mypkg.mymodule -> mypkg/mymodule.py)
parts = fullname.split(".")
module_path = self.base_path.joinpath(*parts)
# 检查是否存在 .py 文件或目录
if module_path.with_suffix(".py").exists():
return importlib.util.spec_from_file_location(
fullname,
str(module_path.with_suffix(".py")),
submodule_search_locations=[str(module_path.parent)]
)
elif module_path.is_dir() and (module_path / "__init__.py").exists():
return importlib.util.spec_from_file_location(
fullname,
str(module_path / "__init__.py"),
submodule_search_locations=[str(module_path)]
)
else:
return None
@@ -0,0 +1,127 @@
import os
import uuid
from loguru import logger
import mimetypes
from datetime import datetime
from deepdata.sql_manager.persistence_atction import TaskInfoPersistence
class FileScanner:
def __init__(self, dataset_id):
self.dataset_id = dataset_id
self.persistence = TaskInfoPersistence()
def get_existing_paths(self):
"""
优化点1:一次性获取所有已存在的文件路径,存为 Set
"""
logger.info("Fetching existing files from DB...")
existing_files = self.persistence.query_existing_files(self.dataset_id)
# fetchall 返回的是 list of tuples [('path1',), ('path2',)]
# 我们将其转换为 set {'path1', 'path2'} 以实现 O(1) 查找速度
existing_set = {row[0] for row in existing_files}
logger.info(f"Found {len(existing_set)} existing files in DB.")
return existing_set
def prepare_file_data(self, sample, file_id):
"""
优化点2:这部分逻辑从原来的 update_file_result 剥离出来
只负责数据组装,不负责插入数据库
"""
file_size = str(sample.get("fileSize"))
file_type = str(sample.get("fileType"))
file_name = str(sample.get("fileName"))
dataset_id = str(sample.get("dataset_id"))
file_path = str(sample.get("filePath"))
create_time = datetime.now()
# 获取最后访问时间,增加异常处理
try:
last_access_time = datetime.fromtimestamp(os.path.getmtime(file_path))
except (FileNotFoundError, OSError):
last_access_time = create_time
# 返回字典,供 executemany 使用
return {
"id": file_id,
"dataset_id": dataset_id,
"file_name": file_name,
"file_path": file_path,
"file_type": file_type,
"file_size": file_size,
"status": "COMPLETED",
"upload_time": create_time,
"last_access_time": last_access_time,
"created_at": create_time,
"updated_at": create_time
}
def scan_and_process(self, root_dir, batch_size=5000):
logger.info(f"Scanning directory: {root_dir}")
# 1. 内存中收集扫描到的所有文件 {path: metadata_dict}
scanned_files_map = {}
for root, dirs, files in os.walk(root_dir):
for file in files:
if file.startswith('.'): continue
full_path = os.path.join(root, file)
# 预先收集元数据
try:
stats = os.stat(full_path)
f_type, _ = mimetypes.guess_type(full_path)
if not f_type: f_type = os.path.splitext(file)[1]
# 构造 sample 格式
scanned_files_map[full_path] = {
"fileSize": stats.st_size,
"fileType": f_type,
"fileName": file,
"dataset_id": self.dataset_id,
"filePath": full_path
}
except OSError:
continue
logger.info(f"Scanned {len(scanned_files_map)} files on disk.")
# 2. 获取数据库中已有的路径
existing_paths = self.get_existing_paths()
# 3. 内存做差集 (Set Difference) -> 找出需要新增的路径
scanned_paths_set = set(scanned_files_map.keys())
new_paths = list(scanned_paths_set - existing_paths)
logger.info(f"Need to insert {len(new_paths)} new files.")
if not new_paths:
logger.info("No new files to insert.")
return
# 4. 准备批量插入的数据
insert_batch = []
total_inserted = 0
for path in new_paths:
sample_data = scanned_files_map[path]
new_file_id = str(uuid.uuid4())
# 调用转换逻辑
record = self.prepare_file_data(sample_data, new_file_id)
insert_batch.append(record)
# 优化点3:分批执行,防止一次性数据包过大导致 SQL 报错
if len(insert_batch) >= batch_size:
self.persistence.batch_insert_files(insert_batch)
total_inserted += len(insert_batch)
logger.info(f"Progress: {total_inserted}/{len(new_paths)} inserted...")
insert_batch = [] # 清空列表
# 插入剩余的数据
if insert_batch:
self.persistence.batch_insert_files(insert_batch)
total_inserted += len(insert_batch)
logger.info(f"Done. Total inserted: {total_inserted}")
@@ -0,0 +1,245 @@
# -*- coding: utf-8 -*-
import os
import re
import sys
import importlib
import subprocess
from pathlib import Path
from types import ModuleType
from loguru import logger
from packaging.version import parse as parse_version
def is_valid_whl_filename(package_name):
"""
验证WHL文件名是否安全(聚焦防范命令注入攻击)
Args:
package_name (str): 要验证的whl文件名
Returns:
bool: True表示安全,False表示包含危险字符
"""
# 禁止路径分隔符
if os.path.sep in package_name or (os.path.altsep and os.path.altsep in package_name):
return False
# 定义危险字符黑名单
dangerous_pattern = re.compile(r"""
[ # 匹配任意以下字符
; & | ` # 命令分隔符
$ () {} <> # 变量展开/命令替换
\] \[ # 特殊符号
! \\ # 历史扩展/转义符
'"*?#\s # 引号/通配符/空格
]
""", re.VERBOSE)
if dangerous_pattern.search(package_name):
return False
return True
class PackageNotFoundError(Exception):
pass
class LazyLoader(ModuleType):
def __init__(self,
package_name,
module_name=None,
whl_path=None,
exact_version=None,
force_reinstall=False
):
"""
:param package_name: WHL包名称中的模块名称部分(一般是模块名称_替换为-)
:param module_name: WHL包安装后,可用于import的模块名称, 当whl包名称和import名称不一致时,填写local_name.
:param whl_path: WHL文件所在目录
:param exact_version: 精确版本要求
:param force_reinstall: 强制重新安装
"""
try:
frame = sys._getframe(1)
self._parent_globals = frame.f_globals
except (AttributeError, ValueError) as e:
logger.error(f"Failed to get stack frame: {e}")
raise RuntimeError("Stack frame retrieval failed") from e
self._module_name = module_name if module_name else package_name
self._package_name = package_name
self.whl_path = whl_path
self.exact_version = exact_version
self.force_reinstall = force_reinstall
self._cached_module = None
# 注册别名到父级命名空间
self._parent_globals[self._module_name] = self
super().__init__(self._module_name)
def __getattr__(self, name):
if self._cached_module is None:
self._cached_module = self._load_module()
return getattr(self._cached_module, name)
def __dir__(self):
return dir(self._load_module())
def _load_module(self):
"""模块加载逻辑"""
if self._cached_module is not None:
return self._cached_module
package_name: str = self._package_name.split('.')[0]
if not is_valid_whl_filename(package_name):
logger.error(f"Invalid package_name: {package_name}")
raise RuntimeError("Invalide package_name, please check it again!")
module_name = self._module_name if self._module_name else package_name.replace("_", "-")
need_install = False
try:
if not self.force_reinstall:
module = importlib.import_module(module_name)
self._cached_module = module
self._register_alias(module)
return module
except ImportError:
need_install = True
if self.force_reinstall:
# 强制安装时的版本检查
installed = self._check_package_exists(package_name)
if installed and self.exact_version:
installed_version = self._get_installed_version(package_name)
if parse_version(installed_version) != parse_version(self.exact_version):
logger.info(f"Version mismatch detected: {installed_version} vs {self.exact_version}")
need_install = True
else:
need_install = True
if need_install:
if self.whl_path is None:
self._pip_install_package_pypi(package_name)
else:
self._pip_install_package_local(package_name)
module = importlib.import_module(module_name)
self._cached_module = module
self._register_alias(module)
else:
# 版本检查通过,无需再次安装
module = importlib.import_module(module_name)
self._cached_module = module
self._register_alias(module)
return self._cached_module
def _register_alias(self, module):
"""注册本地别名 """
self._parent_globals[self._module_name] = module
sys.modules[self._module_name] = module
def _check_package_exists(self, package_name):
"""增强版包检查"""
try:
result = subprocess.run(
[sys.executable, "-m", "pip", "show", package_name],
capture_output=True,
text=True
)
return result.returncode == 0
except subprocess.SubprocessError as e:
logger.error(f"Package check failed: {e}")
return False
def _get_installed_version(self, package_name):
"""获取已安装版本 """
result = subprocess.run(
[sys.executable, "-m", "pip", "show", package_name],
capture_output=True,
text=True
)
for line in result.stdout.split('\n'):
if line.startswith('Version:'):
return line.split()[-1]
raise PackageNotFoundError()
def _pip_install_package_pypi(self, package_name: str):
if self.exact_version:
package_name += f"=={self.exact_version}"
try:
subprocess.check_call([
sys.executable, "-m", "pip", "install", str(package_name)
], stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
logger.info(f"Successfully installed {package_name}")
except subprocess.CalledProcessError as e:
logger.error(f"Installation failed: {e}")
raise RuntimeError(f"Installation failed: {e}") from e
def _pip_install_package_local(self, package_name: str):
"""安装逻辑 """
whl_path = Path(self.whl_path).resolve()
if not whl_path.exists():
raise FileNotFoundError(f"WHL directory not found: {self.whl_path}")
whl_files = list(whl_path.glob(f"{package_name}*.whl"))
if not whl_files:
raise RuntimeError(f"No WHL files found for {package_name}")
# 版本过滤
if self.exact_version:
pattern = re.compile(
rf'^{re.escape(package_name)}-{re.escape(self.exact_version)}-\S*\.whl$',
re.IGNORECASE
)
whl_files = [f for f in whl_files if pattern.match(f.name)]
# 选择最新版本
whl_versions = []
for f in whl_files:
try:
version = self._extract_version(f)
whl_versions.append((f, version))
except ValueError:
continue
if not whl_versions:
raise FileNotFoundError("No valid WHL files")
whl_versions.sort(key=lambda x: x[1], reverse=True)
target_whl = whl_versions[0][0]
# 执行安装
try:
subprocess.check_call([
sys.executable, "-m", "pip", "install",
"--no-index",
f"--find-links={self.whl_path}",
str(target_whl)
], stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
logger.info(f"Successfully installed {target_whl}")
except subprocess.CalledProcessError as e:
logger.error(f"Installation failed: {e}")
raise RuntimeError(f"Installation failed: {e}") from e
def _extract_version(self, filename):
"""版本解析 """
version_pattern = r"(\d+([.]\d+)+([ab]|rc\d+)*([.]post\d+)*([.]dev\d+)*)"
match = re.search(
rf"^{re.escape(self._package_name)}-({version_pattern})",
filename.name,
re.IGNORECASE
)
if not match:
raise ValueError(f"Invalid version format: {filename.name}")
return parse_version(match.group(1))
@@ -0,0 +1,118 @@
# -*- coding: utf-8 -*-
import json
import os
import ssl
from pathlib import Path
from typing import Dict
from urllib.request import Request, urlopen
import requests
import urllib3
from loguru import logger
from deepdata.common.utils import decrypt
class LlmReq:
# 定义常量用于解释错误码
ERRORCODE_INCOMPLETE_CONFIG = 83005
ERRORCODE_INVALID_RESPONSE = 83006
ERRORCODE_SERVICE_UNAVAILABLE = 83007
def __init__(self, url: str = None, header: Dict = None, body: Dict = None, access_type: int = None,
is_https: bool = False, is_certificate: bool = False, certificate_path: Path = None):
self.url = url
self.header = header
self.access_type = access_type
self.is_https = is_https
self.context = self._load_certificate(certificate_path, is_certificate) if is_https else None
self.body = body
if not self.body.get("messages", [])[0].get("content"):
self.body["messages"][0]["content"] = "你好"
def __call__(self, input_str: str) -> str:
outputs = ''
try:
self.body["messages"][0]["content"] = input_str
outputs = self._call_service()
except KeyError as e:
logger.error(f"The body format is not completed, error detail: {e}")
self.body["messages"][0]["content"] = "你好"
return outputs
@staticmethod
def _load_certificate(certificate_path: Path, is_certificate: bool) -> ssl.SSLContext:
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.check_hostname = False
if is_certificate:
context.load_verify_locations(certificate_path)
context.verify_mode = ssl.CERT_REQUIRED
else:
context.verify_mode = ssl.CERT_NONE
return context
@staticmethod
def _pool_manager():
cert_file = os.getenv("RAY_TLS_SERVER_CERT", "/certPersonal/global/identity/global.crt")
key_file = os.getenv("RAY_TLS_SERVER_KEY", "/certPersonal/global/identity/global.key")
ca_crt = os.getenv("RAY_TLS_CA_CERT", "/certPersonal/global/trust/ca.crt")
pwd = os.getenv("RAY_TLS_SERVER_KEY_PASSWORD", "/certPersonal/global/identity/pwd.txt")
key_password = os.getenv("GLOBAL_PWD", None)
if not key_password:
with open(pwd, "r") as f:
key_password = f.read().strip()
key_password = decrypt(key_password)
pool_manager = urllib3.PoolManager(cert_file=cert_file,
key_file=key_file,
key_password=key_password,
cert_reqs='CERT_REQUIRED',
ca_certs=ca_crt,
assert_hostname='edeepdata',
ssl_version='TLSv1_2')
return pool_manager
def _call_service(self):
if not all([self.url, self.header, self.body.get("messages", [])[0].get("content")]):
logger.error("LLM is not configured completely")
raise RuntimeError(self.ERRORCODE_INCOMPLETE_CONFIG, "LLM is not configured completely") from None
if not self.access_type:
try:
pool_manager = self._pool_manager()
response = pool_manager.request(
"POST",
url=self.url,
body=json.dumps(self.body).encode(),
headers=self.header
)
logger.info(f"Response status code: {response.status}")
response_json = json.loads(response.data.decode('utf-8'))
outputs = response_json.get("choices", [])[0].get("message", {}).get("content")
if not outputs:
logger.error("Invalid response format for LLM, missing the 'prompt' key word")
raise RuntimeError(self.ERRORCODE_INVALID_RESPONSE,
"Invalid response format for LLM, missing the 'prompt' key word") from None
return outputs
except Exception as e:
logger.error(f"LLM service is not available, error detail: {e}")
raise RuntimeError(self.ERRORCODE_SERVICE_UNAVAILABLE, "LLM service is not available") from None
if self.access_type:
try:
if self.is_https:
req = Request(url=self.url, data=json.dumps(self.body).encode(), headers=self.header, method="POST")
response_json = urlopen(req, context=self.context).read().decode("utf-8")
response = json.loads(response_json)
else:
response = requests.post(url=self.url, data=json.dumps(self.body), headers=self.header,
stream=False).json()
outputs = response.get("choices", [])[0].get("message", {}).get("content")
if not outputs:
logger.error("Invalid response format for LLM, missing the 'prompt' key word")
raise RuntimeError(self.ERRORCODE_INVALID_RESPONSE,
"Invalid response format for LLM, missing the 'prompt' key word") from None
return outputs
except Exception as e:
logger.error(f"LLM service is not available, error detail: {e}")
raise RuntimeError(self.ERRORCODE_SERVICE_UNAVAILABLE, "LLM service is not available") from None
return None # 确保在所有情况下都返回
@@ -0,0 +1,109 @@
# -*- coding: utf-8 -*-
import sys
import re
import subprocess
from pathlib import Path
from typing import Optional
from loguru import logger
from packaging.version import parse as parse_version, Version
def install_whl(
package_name: str,
whl_path: str,
exact_version: Optional[str] = None,
filename_pattern: Optional[str] = None,
force_reinstall: bool = False
) -> None:
"""
:param package_name: eg: ("zh_core_web_sm")
:param whl_path: WHL file save path
:param exact_version: version number
:param filename_pattern: custom filename pattern for REGEX
:param force_reinstall: which decide to overlap the original number or not (default: False)
"""
whl_dir = Path(whl_path).resolve()
whl_files = _get_whl_files(exact_version, filename_pattern, package_name, whl_dir)
# 语义化版本排序
target_whl = _sort_whl_files(whl_files)
# 安装命令
cmd = [
sys.executable, "-m", "pip", "install",
"--no-index",
f"--find-links={whl_dir}",
str(target_whl)
]
if force_reinstall:
cmd.append("--force-reinstall")
try:
subprocess.check_call(
cmd,
stdout=subprocess.DEVNULL,
stderr=subprocess.STDOUT
)
logger.info(f"Successfully installed {target_whl.name}")
except subprocess.CalledProcessError as e:
error_msg = (
f"Installation failed for {package_name}\n"
f"Possible reasons:\n"
f"1. Missing dependencies in {whl_dir}\n"
f"2. Incompatible Python version\n"
f"3. Platform mismatch (e.g., x86 vs ARM)"
)
raise RuntimeError(error_msg) from e
def _sort_whl_files(whl_files):
whl_versions = []
logger.info(f"[load_offline_module]whl_files: {whl_files}")
for f in whl_files:
try:
version = _extract_version(f)
whl_versions.append((f, version))
except ValueError as e:
logger.warning(f"Skipping invalid file {f.name}: {e}")
continue
if not whl_versions:
raise FileNotFoundError("No valid WHL files with parseable versions")
whl_versions.sort(key=lambda x: x[1], reverse=True)
target_whl = whl_versions[0][0]
return target_whl
def _get_whl_files(exact_version, filename_pattern, package_name, whl_dir):
# 正则表达式
if filename_pattern:
pattern = filename_pattern
else:
if exact_version:
version_part = re.escape(exact_version)
pattern = rf"^{re.escape(package_name)}-{version_part}-\S*\.whl$"
else:
pattern = rf"^{re.escape(package_name)}\S*\.whl$"
regex = re.compile(pattern, re.IGNORECASE)
whl_files = [f for f in whl_dir.glob("*.whl") if regex.match(f.name)]
if not whl_files:
available_files = "\n".join([f.name for f in whl_dir.glob("*.whl")])
raise FileNotFoundError(
f"No matching WHL found for {package_name} in {whl_dir}\n"
f"Available files:\n{available_files}"
)
return whl_files
def _extract_version(filename: Path) -> Version:
"""从文件名提取语义化版本("""
match = re.search(
r"-(\d+([.]\d+)+([ab]|rc\d+)*([.]post\d+)*([.]dev\d+)*)-",
filename.name
)
if not match:
raise ValueError(f"Invalid version format: {filename.name}")
return parse_version(match.group(1))
@@ -0,0 +1,102 @@
# -*- coding: utf-8 -*-
from loguru import logger
class Registry(object):
"""注册器类,用于注册所有的算子."""
def __init__(self, name: str):
self._name = name
self._modules = {}
@property
def name(self):
return self._name
@property
def modules(self):
return self._modules
def list(self):
"""日志打印注册器中所有的算子"""
for m in self._modules.keys():
logger.info(f'{self._name}\t{m}')
def get(self, module_key):
return self._modules.get(module_key, None)
def register_module(self, module_name: str = None, module_cls: type = None, module_path: str = None, force=False):
"""
使用特定的模块名称注册模块
:param module_name: 模块名称
:param module_cls: 模块类定义
:param module_path: 模块所在的路径
:param force: 是否强行覆盖同名模块,默认值为False.
Example:
>>> registry = Registry()
>>> @registry.register_module()
>>> class TextFormatter:
>>> pass
>>> class TextFormatter2:
>>> pass
>>> registry.register_module( module_name='text_formatter2', module_cls=TextFormatter2)
"""
if not (module_name is None or isinstance(module_name, str)):
raise TypeError(f'module_name must be either of None, str,'
f'got {type(module_name)}')
if module_cls is not None:
self._register_module(module_name=module_name,
module_cls=module_cls,
force=force)
return module_cls
elif module_cls is None and isinstance(module_path, str):
self._register_module(module_name=module_name,
module_path=module_path,
force=force)
return module_path
def _register(module_cls):
"""
注册其中module_cls为None是,返回装饰器函数
"""
self._register_module(module_name=module_name,
module_cls=module_cls,
force=force)
return module_cls
return _register
def _register_module(self, module_name=None, module_cls=None, module_path=None, force=False):
"""
注册模块到注册器中.
:param module_name: 模块名称
:param module_cls: 模块类定义
:param force: 是否强行覆盖同名模块,默认值为False.
"""
if module_name is None and module_cls is not None:
module_name = module_cls.__name__
if module_name in self._modules:
if module_cls is not None and module_cls == self._modules[module_name]:
return
if module_path is not None and module_path == self._modules[module_name]:
return
if not force:
raise KeyError(
f'{module_name} is already registered in {self._name}, content: {self.modules.keys()}')
if module_cls is not None:
self._modules[module_name] = module_cls
elif module_path is not None:
self._modules[module_name] = module_path
@@ -0,0 +1,53 @@
# -*- coding: utf-8 -*-
import requests
from typing import Optional, Dict, Any
from starlette.responses import JSONResponse
def http_request(method: str, url: str, data: Optional[Dict[str, Any]] = None):
"""
通用HTTP请求方法
Args:
method: 请求方法 ('GET', 'POST', 'PUT', 'DELETE', 'PATCH')
url: 请求URL
data: 请求数据(JSON格式)
Returns: 格式化响应体
Raises:
requests.exceptions.RequestException: 请求异常
ValueError: 响应状态码错误
"""
success_codes = [200, 201, 202, 204]
# 设置默认请求头
headers = {"Content-Type": "application/json"}
method = method.upper()
try:
# 准备请求参数
request_kwargs = {"url": url, "headers": headers}
# 根据请求方法添加数据
if data is not None:
if method in ["POST", "PUT", "DELETE"]:
request_kwargs["json"] = data
elif method == "GET":
request_kwargs["params"] = data
# 发送请求
response = requests.request(method, **request_kwargs)
# 检查响应状态码
if response.status_code not in success_codes:
raise ValueError(
f"Request failed! Status code: {response.status_code}, "
f"content: {response.text}, "
f"URL: {url}"
)
return response
except requests.exceptions.RequestException as e:
raise requests.exceptions.RequestException(f"Request exception: {str(e)}")
@@ -0,0 +1,171 @@
#!/user/bin/python
import re
from typing import List
from collections import deque
from loguru import logger
class TextSplitter:
"""文本切片"""
# 基于常用标点符号分句,保持句子完整
COMMON_PUNCTUATIONS = [",", "。", "?", "!", ";", ",", "?", "!", ";"]
PUNC_PATTERN = f"[{''.join(COMMON_PUNCTUATIONS)}]"
def __init__(self, max_characters: int, chunk_size: int, chunk_overlap: int):
"""文本切片初始化
Args:
max_characters :文件最大字符,超过截断,-1不处理
chunk_size: 块大小
chunk_overlap: 块重叠度
"""
if chunk_size <= chunk_overlap:
logger.error(f"param chunk_size should larger than chunk_overlap, "
f"current chunk_size: {chunk_size}, chunk_overlap: {chunk_overlap}")
raise Exception(83000, str(ValueError)) from None
self.max_characters = max_characters
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.separators = ["\n\n", "\n"]
@staticmethod
def split_text_by_separator(text: str, separator: str):
"""指定分隔符对文本进行切分,并且切分后的片段需要保留分隔符"""
# 处理一个换行符与两个换行符之间的冲突
if text.startswith("\n\n") and separator == "\n":
chunks = re.split(f"({separator})", text.strip())
chunks[0] = f"\n\n{chunks[0]}"
else:
chunks = re.split(f"({separator})", text)
new_chunks = [chunks[idx] + chunks[idx + 1] for idx in range(1, len(chunks), 2)]
new_chunks = [chunks[0]] + new_chunks
return [chunk for chunk in new_chunks if chunk.strip() != ""]
@staticmethod
def split_sentences(chunk: str):
"""对切片按照标点符号切分成句子,并且保持标点符号不丢失"""
sentences = re.split(TextSplitter.PUNC_PATTERN, chunk)
delimiters = [s for s in chunk if s in TextSplitter.COMMON_PUNCTUATIONS]
restore_chunks = []
for chunk, delimiter in zip(sentences[:-1], delimiters):
restore_chunks.append(chunk + delimiter)
return restore_chunks + [sentences[-1]]
def split_text(self, input_data: str):
if self.max_characters > 0:
logger.info(f"The document characters should be within: {self.max_characters}")
input_data = input_data[:self.max_characters]
logger.info(f"characters of the document: {len(input_data)}")
chunks = self.split_text_recursive(input_data, self.separators)
final_chunks = self.merge_chunks(chunks)
final_chunks = self.split_text_by_chunk_size(final_chunks)
return [chunk.strip() for chunk in final_chunks if chunk]
def split_text_recursive(self, input_data: str, separators: List[str]):
"""对文档按照分隔符优先级进行递归切分:
1. 符合chunk_size要求的切片不再切分。
2. 大于chunk_size要求的切片,继续进行递归切分。
Args:
input_data: 输入文本
separators: 分隔符
Returns:
List[str]: 切分后的文本片段
"""
chunks = []
cur_separator = ""
next_separators = []
for idx, sep in enumerate(separators):
sep = re.escape(sep)
if re.search(sep, input_data.strip()):
cur_separator = sep
next_separators = separators[idx + 1:]
break
if not cur_separator:
return [input_data]
else:
cur_chunks = TextSplitter.split_text_by_separator(input_data, cur_separator)
for chunk in cur_chunks:
if len(chunk.strip()) <= self.chunk_size:
chunks.append(chunk)
else:
if not next_separators:
chunks.append(chunk)
else:
next_chunks = self.split_text_recursive(chunk, next_separators)
chunks.extend(next_chunks)
return chunks
def merge_chunks(self, chunks: List[str]):
"""对切分后的文本片段进行合并,合并过程考虑overlap"""
final_chunks = []
idx = 0
while idx < len(chunks):
if len(chunks[idx]) >= self.chunk_size:
final_chunks.append(chunks[idx])
idx += 1
continue
merge_idxes = self.get_merge_idxes(idx, chunks)
content = ""
for inner_idx in merge_idxes:
content += chunks[inner_idx]
final_chunks.append(content)
idx = merge_idxes[-1] + 1
return final_chunks
def get_merge_idxes(self, cur_idx: int, chunks: List[str]):
"""获取可以合并的分片index,前向尽可能满足overlap,后向尽可能满足chunk_size"""
idxes = deque([cur_idx])
overlap_idx = cur_idx - 1
cur_len = len(chunks[cur_idx])
cur_idx += 1
# 获取overlap的index
over_lap_len = 0
while overlap_idx >= 0:
over_lap_len += len(chunks[overlap_idx])
if over_lap_len > self.chunk_overlap or (cur_len + over_lap_len) > self.chunk_size:
over_lap_len -= len(chunks[overlap_idx])
break
idxes.appendleft(overlap_idx)
overlap_idx -= 1
cur_len += over_lap_len
# 获取merge的index
while cur_idx < len(chunks):
cur_len += len(chunks[cur_idx])
if cur_len > self.chunk_size:
break
idxes.append(cur_idx)
cur_idx += 1
return idxes
def split_chunks(self, chunks: List[str]):
"""对超过`chunk_size`限制的切片进行截断,过程中需要考虑overlap参数"""
final_chunks = []
for chunk in chunks:
if len(chunk) <= self.chunk_size:
final_chunks.append(chunk)
else:
start = 0
end = self.chunk_size
while end < len(chunk):
final_chunks.append(chunk[start: end])
start += self.chunk_size - self.chunk_overlap
end = start + self.chunk_size
final_chunks.append(chunk[start:])
return final_chunks
def split_text_by_chunk_size(self, chunks: List[str]):
"""对切片后超长的文本块进行二次切分,使用截断,并考虑overlap"""
final_chunks = []
for chunk in chunks:
if len(chunk) <= self.chunk_size:
final_chunks.append(chunk)
continue
sentences = TextSplitter.split_sentences(chunk)
sub_chunks = self.merge_chunks(sentences)
final_chunks.extend(self.split_chunks(sub_chunks))
return final_chunks
@@ -0,0 +1,677 @@
# -*- coding: utf-8 -*-
import json
import mimetypes
import os
import time
import traceback
import uuid
from pathlib import Path
from typing import List, Dict, Any, Tuple
import cv2
import numpy as np
from loguru import logger
from unstructured.partition.auto import partition
from deepdata.common.error_code import ERROR_CODE_TABLE, UNKNOWN_ERROR_CODE
from deepdata.common.utils.llm_request import LlmReq
from deepdata.common.utils.registry import Registry
from deepdata.common.utils import check_valid_path
from deepdata.core.constant import Fields
from deepdata.sql_manager.persistence_atction import TaskInfoPersistence
OPERATORS = Registry("Operators")
FAILED_STATUS = "FAILED"
SUCCESS_STATUS = "COMPLETED"
def get_exception_info(e):
exc_type = type(e).__name__ # 异常类型(如 'ZeroDivisionError')
exc_msg = str(e) # 异常原因(如 'division by zero')
# 提取详细的堆栈信息
tb = traceback.extract_tb(e.__traceback__) # 解析 traceback 对象
error_line = tb[-1].lineno # 错误发生的行号
error_file = tb[-1].filename # 错误发生的文件名
code_snippet = tb[-1].line # 错误行的代码
# 组合输出信息
error_info = (
f"错误类型: {exc_type}\n"
f"错误原因: {exc_msg}\n"
f"文件名: {error_file}\n"
f"行号: {error_line}\n"
f"代码行: {code_snippet}"
)
return error_info
class BaseOp:
"""
所有算子类的父类
"""
use_model = False
custom_ops = False
def __init__(self, *args, **kwargs):
self.accelerator = kwargs.get("accelerator", "cpu")
self.is_last_op = kwargs.get("is_last_op", False)
self.is_first_op = kwargs.get("is_first_op", False)
self._name = kwargs.get("op_name", None)
self.infer_model = None
self.text_key = kwargs.get("text_key", "text")
self.data_key = kwargs.get("data_key", "data")
self.image_key = kwargs.get("image_key", "image")
self.video_key = kwargs.get("video_key", "video")
self.audio_key = kwargs.get("audio_key", "audio")
self.filename_key = kwargs.get("fileName_key", "fileName")
self.filetype_key = kwargs.get("fileType_key", "fileType")
self.fileid_key = kwargs.get("fileId_key", "fileId")
self.filepath_key = kwargs.get("filePath_key", "filePath")
self.filesize_key = kwargs.get("fileSize_key", "fileSize")
self.export_path_key = kwargs.get("export_path_key", "export_path")
self.ext_params_key = kwargs.get("ext_params_key", "ext_params")
self.target_type_key = kwargs.get("target_type_key", "target_type")
@property
def name(self):
if self._name:
return self._name
else:
return "UnknownOp"
@staticmethod
def is_npu_available():
try:
import torch_npu
return torch_npu.npu.is_available()
except ImportError as e:
logger.warning("Import torch_npu failed.")
return False
@staticmethod
def update_kwargs(
sample: Dict[str, Any], not_update_keys=("text", "data", "meta")
) -> Dict:
"""获取sample_data中文件相关的信息"""
res = {}
for k, v in sample.items():
if k not in not_update_keys:
res[k] = v
return res
@staticmethod
def _get_error_info(e: BaseException) -> Tuple[str, str]:
error_code = UNKNOWN_ERROR_CODE
exc_info = get_exception_info(e)
for exc_type in type(e).__mro__:
if exc_type in ERROR_CODE_TABLE.keys():
error_code = ERROR_CODE_TABLE[exc_type]
break
return error_code, exc_info
def use_npu(self):
"""确认算子是否可以使用npu"""
return self.accelerator == "npu" and self.is_npu_available()
def get_model(self, *args, **kwargs):
if self.infer_model is None and self.use_model:
return self.init_model(*args, **kwargs)
else:
logger.info(
f"Op named {self.name} get infer model Failed. please "
f" check Attribute self.use_model: {self.use_model} or model has been initialized!"
)
return self.infer_model
def init_model(self, *args, **kwargs):
"""执行函数(子类实现)"""
raise NotImplementedError(
"This is in BaseOp, plese re-define this method in Sub-classes"
)
def fill_sample_params(self, sample: Dict[str, Any], **kwargs):
if not sample.get(self.text_key, None):
sample[self.text_key] = ""
if not sample.get(self.data_key, None):
sample[self.data_key] = b""
if not sample[self.data_key] and not sample[self.text_key]:
sample.update(kwargs)
def create_failure_sample(
self, sample: Dict[str, Any], op_name, excp: BaseException
):
sample["execute_result"] = False
error_code, exc_info = self._get_error_info(excp)
failed_reason = {
"op_name": op_name,
"error_code": error_code,
"reason": exc_info,
}
sample["failed_reason"] = failed_reason
def read_file(self, sample):
filepath = sample[self.filepath_key]
filetype = sample[self.filetype_key]
if filetype in ["ppt", "pptx", "docx", "doc", "xlsx", "csv", "md", "pdf"]:
elements = partition(filename=filepath)
sample[self.text_key] = "\n\n".join([str(el) for el in elements])
sample[self.data_key] = b""
elif filetype in ["txt", "md", "markdown", "xml", "html", "json", "jsonl"]:
with open(filepath, "rb") as f:
content = f.read()
sample[self.text_key] = content.decode("utf-8-sig").replace(
"\r\n", "\n"
)
sample[self.data_key] = b""
elif filetype in ["jpg", "jpeg", "png", "bmp"]:
image_np = cv2.imdecode(np.fromfile(filepath, dtype=np.uint8), -1)
if image_np.size:
data = cv2.imencode(f".{filetype}", image_np)[1]
image_bytes = data.tobytes()
sample[self.data_key] = image_bytes
sample[self.text_key] = ""
return sample
def read_file_first(self, sample):
if self.is_first_op:
self.read_file(sample)
def convert_to_dj(self, sample):
filepath = sample[self.filepath_key]
mime_type, _ = mimetypes.guess_type(filepath)
file_type = None
if mime_type:
file_type = mime_type.split("/")[0]
if file_type == "text":
return self.read_file(sample)
elif file_type == "image":
sample["text"] = ""
sample["data"] = b""
sample["images"] = [filepath]
elif file_type == "audio":
sample["text"] = ""
sample["data"] = b""
sample["audios"] = [filepath]
elif file_type == "video":
sample["text"] = ""
sample["data"] = b""
sample["videos"] = [filepath]
return sample
@staticmethod
def save_file_and_db(sample):
if FileExporter().execute(sample):
TaskInfoPersistence().persistence_task_info(sample)
return sample
class Mapper(BaseOp):
def __init__(self, *args, **kwargs):
super(Mapper, self).__init__(*args, **kwargs)
def __call__(self, sample: Dict[str, Any], **kwargs):
# 该算子前已有算子执行该文件失败
if sample.get(Fields.result) is False:
return sample
self.fill_sample_params(sample, **kwargs)
execute_status = FAILED_STATUS
try:
sample = self.execute(sample)
execute_status = SUCCESS_STATUS
except Exception as e:
# 算子执行失败,记录文件执行信息到数据库,并更该文件执行结果状态
self.create_failure_sample(sample, self.name, e)
logger.error(
f"Ops named {self.name} map failed, Error Info: \n"
f"{str(get_exception_info(e))}"
)
sample["execute_status"] = execute_status
sample[self.filesize_key] = "0"
sample[self.filetype_key] = ""
sample["execute_result"] = False
TaskInfoPersistence().update_task_result(sample)
# 不抛出异常,跳过当前文件继续处理下一个文件
return sample
sample["execute_status"] = execute_status
# 加载文件成功执行信息到数据库
if self.is_last_op:
# 文件无内容会被过滤
if sample[self.text_key] == "" and sample[self.data_key] == b"":
task_info = TaskInfoPersistence()
sample[self.filesize_key] = "0"
sample[self.filetype_key] = ""
task_info.update_task_result(sample)
return sample
else:
self.save_file_and_db(sample)
return sample
def execute(self, sample: Dict[str, Any]) -> Dict[str, Any]:
"""执行函数(子类实现)"""
raise NotImplementedError(
"This is in Mapper Class, plese re-define this method in Sub-classes"
)
class Slicer(BaseOp):
def __init__(self, *args, **kwargs):
super(Slicer, self).__init__(*args, **kwargs)
self.target_file_type = None
def __call__(self, sample: Dict[str, Any], **kwargs):
# 该算子前已有算子执行该文件失败
if sample.get(Fields.result) is False:
return sample
self.fill_sample_params(sample, **kwargs)
sample_list = []
execute_status = FAILED_STATUS
try:
sample_list = self.execute(sample)
execute_status = SUCCESS_STATUS
except Exception as e:
# 算子执行失败,记录文件执行信息到数据库,并更该文件执行结果状态
self.create_failure_sample(sample, self.name, e)
self.load_sample_to_sample(sample, sample_list)
logger.error(
f"Ops named {self.name} map failed, Error Info: \n"
f"{str(get_exception_info(e))}"
)
sample["execute_status"] = execute_status
sample[self.filesize_key] = "0"
sample[self.filetype_key] = ""
TaskInfoPersistence().update_task_result(sample)
return [sample]
self.load_sample_to_sample(sample, sample_list)
sample["execute_status"] = execute_status
# 加载文件成功执行信息到数据库
if self.is_last_op:
self.save_file_and_db(sample)
return [sample]
@staticmethod
def load_sample_to_sample(sample: Dict, sample_list: List[Dict]):
"""使用sample中的k-v更新sample"""
for sample_i in sample_list:
for k, v in sample_i.items():
sample[k] = v
if not sample.get("fileNum", None):
sample["fileNum"] = 1
def execute(self, sample: Dict[str, Any]) -> List[Dict]:
"""执行函数(子类实现)"""
raise NotImplementedError(
"This is in Mapper Class, plese re-define this method in Sub-classes"
)
def save_patch_sample(self, sample: Dict[str, Any], patch_no, save_format="text"):
if save_format == "text":
target_file_type = "txt"
elif save_format == "image":
target_file_type = "png"
else:
target_file_type = None
raise RuntimeError(f"target file type is {target_file_type}!")
if self.target_file_type:
target_file_type = self.target_file_type
save_path = self.get_save_path(sample, patch_no, target_file_type)
self.save_file(sample, save_path)
def get_save_path(self, sample: Dict[str, Any], patch_no, target_type) -> str:
export_path = os.path.abspath(sample[self.export_path_key])
logger.info(f"export path: {export_path}.")
base_file_name, _ = os.path.splitext(sample[self.filename_key])
file_id = str(sample[self.fileid_key])
new_file_name = file_id + "_" + str(patch_no) + "." + target_type
logger.info(
f"base_file_name: {base_file_name}, new file name: {new_file_name}."
)
if not check_valid_path(export_path):
os.makedirs(export_path, exist_ok=True)
res = os.path.join(export_path, new_file_name)
return res
def save_file(self, sample, save_path):
# 以二进制格式保存文件
file_sample = (
sample[self.text_key].encode("utf-8")
if sample[self.text_key]
else sample[self.data_key]
)
with open(save_path, "wb") as f:
f.write(file_sample)
os.chmod(save_path, 0o640)
try:
parent_dir = os.path.dirname(save_path)
os.chmod(parent_dir, 0o770)
except Exception as e:
logger.warning("Failed to modify the permission on the parent_dir.")
logger.info(f"patch sample has been save to {save_path}.")
class Filter(BaseOp):
def __init__(self, *args, **kwargs):
super(Filter, self).__init__(*args, **kwargs)
def __call__(self, sample: Dict[str, Any], **kwargs):
# 该算子前已有算子执行该文件失败
if sample.get(Fields.result) is False:
return sample
self.fill_sample_params(sample, **kwargs)
execute_status = FAILED_STATUS
try:
sample = self.execute(sample)
execute_status = SUCCESS_STATUS
except Exception as e:
# 如果filter算子过滤失败, 不保留文件, 并记录文件执行信息到数据库
self.create_failure_sample(sample, self.name, e)
return False
sample["execute_status"] = execute_status
# 文件无内容会被过滤
if sample[self.text_key] == "" and sample[self.data_key] == b"":
task_info = TaskInfoPersistence()
sample[self.filesize_key] = "0"
sample[self.filetype_key] = ""
task_info.update_task_result(sample)
return False
# 加载文件成功执行信息到数据库
if self.is_last_op:
self.save_file_and_db(sample)
return True
def execute(self, sample: Dict[str, Any]) -> Dict[str, Any]:
"""执行函数(子类实现)"""
raise NotImplementedError(
"This is in Filter Class, plese re-define this method in Sub-classes"
)
class LLM(Mapper):
def __init__(self, *args, **kwargs):
super(LLM, self).__init__(*args, **kwargs)
self.llm = self.get_llm(*args, **kwargs)
self.prompt_template = None
self.target_file_type = None
def execute(self, sample: Dict[str, Any]) -> Dict[str, Any]:
"""执行函数(子类实现)"""
raise NotImplementedError(
"This is in LLM Class, plese re-define this method in Sub-classes"
)
@staticmethod
def get_llm(*args, **kwargs):
url = kwargs.get("LLMUrl", "")
header = kwargs.get("LLMHeaders", {"Content-type": "application/json"})
body = kwargs.get("LLMBody", {})
access_type = kwargs.get("accessType", False)
is_https = kwargs.get("isHttps", False)
is_certificate = kwargs.get("isCertificate", False)
certificate_path = kwargs.get("certificatePath", None)
return LlmReq(
url=url,
header=header,
body=body,
access_type=access_type,
is_https=is_https,
is_certificate=is_certificate,
certificate_path=certificate_path,
)
def build_llm_prompt(self, *args, **kwargs):
"""执行函数(子类实现)"""
raise NotImplementedError(
"This is in LLM Class, plese re-define this method in Sub-classes"
)
def save_sample(self, object_list: List, sample: Dict[str, Any]):
if self.target_file_type:
target_file_type = self.target_file_type
else:
target_file_type = "jsonl"
save_path = self.get_save_path(sample, target_file_type)
self.save_json_file(object_list, save_path)
def get_save_path(self, sample: Dict[str, Any], target_type) -> str:
export_path = os.path.abspath(sample[self.export_path_key])
logger.info(f"export path: {export_path}.")
base_file_name, _ = os.path.splitext(sample[self.filename_key])
file_id = str(sample[self.fileid_key])
new_file_name = file_id + "." + target_type
logger.info(
f"base_file_name: {base_file_name}, new file name: {new_file_name}."
)
if not check_valid_path(export_path):
os.makedirs(export_path, exist_ok=True)
res = os.path.join(export_path, new_file_name)
return res
@staticmethod
def save_json_file(object_list: List, save_path):
if len(object_list) == 0:
logger.warning(
"Please check the param: object_list, which has length equal to 0."
)
return
try:
with open(save_path, "w", encoding="utf-8") as f:
for item in object_list:
json_str = json.dumps(item, ensure_ascii=False)
f.write(json_str + "\n")
os.chmod(save_path, 0o640)
try:
parent_dir = os.path.dirname(save_path)
os.chmod(parent_dir, 0o770)
except Exception as e:
logger.warning("Failed to modify the permission on the parent_dir.")
except Exception as e:
raise RuntimeError(
f"Save jsonl file Failed!, save_path: {save_path}."
) from e
logger.info(f"LLM output has been save to {save_path}.")
class FileExporter(BaseOp):
"""把输入的json文件流抽取为txt"""
def __init__(self, *args, **kwargs):
super(FileExporter, self).__init__(*args, **kwargs)
self.last_ops = True
self.text_support_ext = kwargs.get(
"text_support_ext",
[
"txt",
"html",
"md",
"markdown",
"xlsx",
"xls",
"csv",
"pptx",
"ppt",
"xml",
"json",
"doc",
"docx",
"pdf",
],
)
self.data_support_ext = kwargs.get(
"data_support_ext", ["jpg", "jpeg", "png", "bmp"]
)
self.medical_support_ext = kwargs.get(
"medical_support_ext", ["svs", "tif", "tiff"]
)
def execute(self, sample: Dict[str, Any]):
file_name = sample[self.filename_key]
file_type = sample[self.filetype_key]
try:
start = time.time()
save_path = ""
if file_type in self.text_support_ext:
sample, save_path = self.get_textfile_handler(sample)
elif file_type in self.data_support_ext:
sample, save_path = self.get_datafile_handler(sample)
elif file_type in self.medical_support_ext:
sample, save_path = self.get_medicalfile_handler(sample)
if sample[self.text_key] == "" and sample[self.data_key] == b"":
if sample.get("executor") == "datajuicer":
return True
sample[self.filesize_key] = "0"
return False
if save_path:
save_path = self.save_file(sample, save_path)
sample[self.text_key] = ""
sample[self.data_key] = b""
sample[Fields.result] = True
file_type = save_path.split(".")[-1]
sample[self.filetype_key] = file_type
file_name = os.path.basename(save_path)
base_name, _ = os.path.splitext(file_name)
new_file_name = base_name + "." + file_type
sample[self.filename_key] = new_file_name
sample[self.filepath_key] = save_path
file_size = os.path.getsize(save_path)
sample[self.filesize_key] = f"{file_size}"
logger.info(f"origin file named {file_name} has been save to {save_path}")
logger.info(
f"fileName: {sample[self.filename_key]}, "
f"method: FileExporter costs {time.time() - start:.6f} s"
)
except UnicodeDecodeError as err:
logger.error(
f"fileName: {sample[self.filename_key]}, "
f"method: FileExporter causes decode error: {err}"
)
raise
return True
def get_save_path(self, sample: Dict[str, Any], target_type):
export_path = os.path.abspath(sample[self.export_path_key])
file_name = sample[self.filename_key]
new_file_name = os.path.splitext(file_name)[0] + "." + target_type
if not check_valid_path(export_path):
os.makedirs(export_path, exist_ok=True)
return os.path.join(export_path, new_file_name)
def get_textfile_handler(self, sample: Dict[str, Any]):
target_type = sample.get("target_type", None)
# target_type存在则保存为扫描件, docx格式
if target_type:
sample = self._get_from_text_or_data(sample)
save_path = self.get_save_path(sample, target_type)
# 不存在则保存为txt文件,正常文本清洗
else:
sample = self._get_from_text(sample)
save_path = self.get_save_path(sample, "txt")
return sample, save_path
def get_datafile_handler(self, sample: Dict[str, Any]):
target_type = sample.get("target_type", None)
# target_type存在, 图转文保存为target_type,markdown格式
if target_type:
sample = self._get_from_text_or_data(sample)
save_path = self.get_save_path(sample, target_type)
# 不存在则保存为原本图片文件格式,正常图片清洗
else:
sample = self._get_from_text_or_data(sample)
save_path = self.get_save_path(sample, sample[self.filetype_key])
return sample, save_path
def get_medicalfile_handler(self, sample: Dict[str, Any]):
target_type = "png"
sample = self._get_from_data(sample)
save_path = self.get_save_path(sample, target_type)
return sample, save_path
def save_file(self, sample, save_path):
# 以二进制格式保存文件
file_sample = (
sample[self.text_key].encode("utf-8")
if sample[self.text_key]
else sample[self.data_key]
)
path_obj = Path(save_path).resolve()
parent_dir = path_obj.parent
stem = path_obj.stem # 文件名不含后缀
suffix = path_obj.suffix # 后缀 (.txt)
counter = 0
current_path = path_obj
while True:
try:
# x 模式保证:如果文件存在则报错,如果不存在则创建。
# 这个检查+创建的过程是操作系统级的原子操作,没有竞态条件。
with open(current_path, "xb") as f:
f.write(file_sample)
break
except FileExistsError:
# 文件已存在(被其他线程/进程抢占),更新文件名重试
counter += 1
new_filename = f"{stem}_{counter}{suffix}"
current_path = parent_dir / new_filename
os.chmod(parent_dir, 0o770)
os.chmod(current_path, 0o640)
return str(current_path)
def _get_from_data(self, sample: Dict[str, Any]) -> Dict[str, Any]:
sample[self.data_key] = bytes(sample[self.data_key])
sample[self.text_key] = ""
return sample
def _get_from_text(self, sample: Dict[str, Any]) -> Dict[str, Any]:
sample[self.data_key] = b""
sample[self.text_key] = str(sample[self.text_key])
return sample
def _get_from_text_or_data(self, sample: Dict[str, Any]) -> Dict[str, Any]:
if (
sample.get(self.data_key) is not None
and sample[self.data_key] != b""
and sample[self.data_key] != ""
):
return self._get_from_data(sample)
else:
return self._get_from_text(sample)
@staticmethod
def _get_uuid():
return str(uuid.uuid4())
@@ -0,0 +1,8 @@
# -*- coding: utf-8 -*-
class Fields(object):
result = 'execute_result'
instance_id = 'instance_id'
export_path = 'export_path'
@@ -0,0 +1,213 @@
# -*- coding: utf-8 -*-
from __future__ import annotations
import os
import re
import importlib
import sys
import uuid
from abc import ABC, abstractmethod
import pyarrow as pa
from enum import Enum
from loguru import logger
from ray import data as rd
from deepdata.core.base_op import Filter, Mapper, Slicer
from deepdata.core.constant import Fields
from deepdata.core.base_op import OPERATORS, BaseOp
from core.base_op import Filter as RELATIVE_Filter, Mapper as RELATIVE_Mapper, Slicer as RELATIVE_Slicer
rd.DataContext.get_current().enable_progress_bars = False
class Formatters(Enum):
"""
抽取算子和落盘算子枚举类
"""
FILE_EXPORTER = "FileExporter"
IMG_FORMATTER = "ImgFormatter"
OCR_FORMATTER = "OcrFormatter"
PDF_CPU_FORMATTER = "PdfCpuFormatter"
SLID_FORMATTER = "SlideFormatter"
TEXT_FORMATTER = "TextFormatter"
WORD_FORMATTER = "WordFormatter"
UNION_FORMATTER = "UnionFormatter"
ONNX_FORMATTER = "OnnxImg2TextFormatter"
@classmethod
def is_member(cls, op_name):
return op_name in cls._value2member_map_
class BasicDataset(ABC):
@abstractmethod
def process(
self,
cfg_process,
*,
exporter=None,
checkpointer=None
) -> BasicDataset:
pass
def preprocess_dataset(dataset: rd.Dataset, cfg) -> rd.Dataset:
columns = dataset.columns()
new_column_names = [getattr(Fields, attr_name)
for attr_name in vars(Fields)
if attr_name not in columns and not attr_name.startswith('__')]
def process_batch_arrow(table: pa.Table, names_list=None) -> pa.Table:
name2value_table = {
Fields.instance_id: cfg.instance_id,
Fields.export_path: cfg.export_path
}
for column_name in names_list:
if column_name in name2value_table.keys():
new_column_data = [name2value_table[column_name] for _ in range(len(table))]
else:
new_column_data = [None for _ in range(len(table))]
table = table.append_column(column_name, [new_column_data])
return table
if new_column_names:
dataset = dataset.map_batches(process_batch_arrow,
fn_kwargs={"names_list": new_column_names},
num_cpus=0.05,
batch_format='pyarrow')
return dataset
class RayDataset(BasicDataset):
def __init__(self,
dataset: rd.Dataset,
cfg=None) -> None:
self.onnx_ops_name = ["OnnxImg2TextFormatter", "OnnxImageContentFilter"]
self.npu_ops_name = ["Img2TextFormatter", "ImageContentFilter"]
self.data = preprocess_dataset(dataset, cfg)
def process(self,
cfg_process,
*,
exporter=None,
checkpointer=None,
**kwargs) -> BasicDataset:
# 从注册器加载类
operators_cls_list = []
init_kwargs_list = []
for index, process in enumerate(cfg_process):
op_name, init_kwargs = list(process.items())[0]
init_kwargs = {} if not init_kwargs else init_kwargs
init_kwargs.update({'op_name': op_name})
# 加载Ops module
temp_ops = self.load_ops_module(op_name)
operators_cls_list.append(temp_ops)
if index == 0:
init_kwargs["is_first_op"] = True
if index == len(cfg_process) - 1:
init_kwargs["is_last_op"] = True
init_kwargs["instance_id"] = kwargs.get("instance_id", str(uuid.uuid4()))
init_kwargs_list.append(init_kwargs)
for cls_id, operators_cls in enumerate(operators_cls_list):
self._run_single_op(operators_cls, init_kwargs_list[cls_id], **kwargs)
return self
def load_ops_module(self, op_name):
'''
加载算子模块
:param op_name: 算子名称
:return: 算子对象
'''
# Validate op_name to prevent path traversal / code injection (CodeQL / FCE)
if not op_name or not re.match(r'^[A-Za-z_][A-Za-z0-9_]*$', op_name):
raise ValueError(f"Invalid operator name: {op_name}")
parent_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "ops")
if parent_dir not in sys.path:
sys.path.insert(0, parent_dir)
registry_content = OPERATORS.modules.get(op_name)
if registry_content is None:
from core.base_op import OPERATORS as RELATIVE_OPERATORS
registry_content = RELATIVE_OPERATORS.modules.get(op_name)
if isinstance(registry_content, str):
# Validate registry_content is a safe dotted module path (no path separators)
if not re.match(r'^[A-Za-z_][A-Za-z0-9_.]*$', registry_content):
raise ValueError(f"Invalid module path for operator {op_name}: {registry_content}")
# registry_content是module的路径
submodule = importlib.import_module(registry_content)
res = getattr(submodule, op_name, None)
if res is None:
raise ImportError(f"Import Ops module {op_name} Failed.")
else:
logger.info(f"Import Ops module {op_name} Success.")
elif isinstance(registry_content, type) and issubclass(registry_content, BaseOp):
# registry_content是module本身
res = registry_content
else:
res = None
return res
def _run_single_op(self, operators_cls, init_kwargs, **kwargs):
max_actor_nums = os.getenv("MAX_ACTOR_NUMS", "20")
resources = {}
if init_kwargs.get("npu", 0) > 0:
resources["npu"] = init_kwargs.get("npu")
if init_kwargs.get("arch", "arm").startswith("x86"):
resources["arch"] = "x86"
cpu = init_kwargs.get("cpu", 0.05)
memory = init_kwargs.get("memory", None)
kwargs.update({"ext_params": {}, "failed_reason": {}, "target_type": None})
try:
if issubclass(operators_cls, (Mapper, RELATIVE_Mapper)):
self.data = self.data.map(operators_cls,
fn_constructor_kwargs=init_kwargs,
fn_kwargs=kwargs,
resources=resources,
num_cpus=cpu,
memory=memory,
compute=rd.ActorPoolStrategy(min_size=1,
max_size=int(max_actor_nums)))
elif issubclass(operators_cls, (Slicer, RELATIVE_Slicer)):
self.data = self.data.flat_map(operators_cls,
fn_constructor_kwargs=init_kwargs,
fn_kwargs=kwargs,
resources=resources,
num_cpus=cpu,
memory=memory,
compute=rd.ActorPoolStrategy(min_size=1,
max_size=int(max_actor_nums)))
elif issubclass(operators_cls, (Filter, RELATIVE_Filter)):
self.data = self.data.filter(operators_cls,
fn_constructor_kwargs=init_kwargs,
fn_kwargs=kwargs,
resources=resources,
num_cpus=cpu,
memory=memory,
compute=rd.ActorPoolStrategy(min_size=1,
max_size=int(max_actor_nums)))
else:
logger.error(
'Ray executor only support Filter, Mapper and Slicer OPs for now')
raise NotImplementedError
except Exception as e:
logger.error(e)
raise Exception("Error! Ops Details:") from e
@@ -0,0 +1,252 @@
import os
import re
from typing import Optional, Dict, Any, List
import uvicorn
import yaml
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from jsonargparse import ArgumentParser
from loguru import logger
from pydantic import BaseModel
from deepdata.common.error_code import ErrorCode
from deepdata.scheduler import cmd_scheduler
from deepdata.scheduler import func_scheduler
from deepdata.scheduler import ray_job_scheduler
from deepdata.wrappers import WRAPPERS
# 日志配置
LOG_DIR = "/var/log/deepdata/runtime"
os.makedirs(LOG_DIR, exist_ok=True)
LOG_ROTATION_BACKUP_COUNT = int(os.getenv("LOG_ROTATION_BACKUP_COUNT", "30"))
def _normalize_log_rotation_size(value: str) -> str:
match = re.match(r"^\s*(\d+)\s*([KMG]?B?)?\s*$", value, re.IGNORECASE)
if not match:
return "100 MB"
amount = match.group(1)
unit = (match.group(2) or "B").upper()
if unit in {"K", "KB"}:
return f"{amount} KB"
if unit in {"M", "MB"}:
return f"{amount} MB"
if unit in {"G", "GB"}:
return f"{amount} GB"
return amount
LOG_ROTATION_MAX_SIZE = _normalize_log_rotation_size(os.getenv("LOG_ROTATION_MAX_SIZE", "100MB"))
logger.add(
f"{LOG_DIR}/runtime.log",
format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {name}:{function}:{line} - {message}",
level="DEBUG",
enqueue=True,
rotation=LOG_ROTATION_MAX_SIZE,
retention=LOG_ROTATION_BACKUP_COUNT,
)
app = FastAPI()
class APIException(Exception):
"""自定义API异常"""
def __init__(self, error_code: ErrorCode, detail: Optional[str] = None,
extra_data: Optional[Dict] = None):
self.error_code = error_code
self.detail = detail or error_code.value[1]
self.code = error_code.value[0]
self.extra_data = extra_data
super().__init__(self.detail)
def to_dict(self) -> Dict[str, Any]:
result = {
"code": self.code,
"message": self.detail,
"success": False
}
if self.extra_data:
result["data"] = self.extra_data
return result
@app.exception_handler(APIException)
async def api_exception_handler(request: Request, exc: APIException):
return JSONResponse(
status_code=200, # 业务错误返回 200,错误信息在响应体中
content=exc.to_dict()
)
class QueryTaskRequest(BaseModel):
task_ids: List[str]
# UUID pattern for task_id validation (prevents path traversal)
_TASK_ID_PATTERN = re.compile(
r'^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$'
)
def _validate_task_id(task_id: str) -> None:
"""Validate task_id is a proper UUID to prevent path traversal (CodeQL)."""
if not _TASK_ID_PATTERN.match(task_id):
raise APIException(ErrorCode.PARAM_ERROR, detail=f"Invalid task_id format: {task_id}")
@app.post("/api/task/list")
async def query_task_info(request: QueryTaskRequest):
try:
return [{task_id: cmd_scheduler.get_task_status(task_id)} for task_id in request.task_ids]
except Exception as e:
raise APIException(ErrorCode.UNKNOWN_ERROR)
class SubmitTaskRequest(BaseModel):
retry_count: int = 0
@app.post("/api/task/{task_id}/submit")
async def submit_task(task_id, request: SubmitTaskRequest = None):
_validate_task_id(task_id)
retry_count = request.retry_count if request else 0
config_path = f"/flow/{task_id}/process.yaml"
logger.info(f"Start submitting job with retry_count={retry_count}...")
dataset_path = get_from_cfg(task_id, "dataset_path")
'''
if not check_valid_path(dataset_path):
logger.error(f"dataset_path is not existed! please check this path.")
raise APIException(ErrorCode.FILE_NOT_FOUND_ERROR)
'''
if dataset_path is not None and not check_valid_path(dataset_path):
logger.error(f"dataset_path is not existed! please check this path.")
raise APIException(ErrorCode.FILE_NOT_FOUND_ERROR)
try:
executor_type = get_from_cfg(task_id, "executor_type")
await WRAPPERS.get(executor_type).submit(task_id, config_path, retry_count)
except Exception as e:
logger.error(f"Error happens during submitting task. Error Info following: {e}")
raise APIException(ErrorCode.SUBMIT_TASK_ERROR)
logger.info(f"task id: {task_id} has been submitted.")
success_json_info = JSONResponse(
content={"status": "Success", "message": f"{task_id} has been submitted"},
status_code=200
)
return success_json_info
@app.get("/api/task/{task_id}/status")
async def get_task_status(task_id):
"""Get task execution status from the scheduler"""
_validate_task_id(task_id)
try:
executor_type = get_from_cfg(task_id, "executor_type", default="deepdata")
if executor_type == "ray" or executor_type == "deepdata":
result = ray_job_scheduler.get_task_status(task_id)
else:
result = cmd_scheduler.get_task_status(task_id)
if result is None:
return JSONResponse(
content={"status": "NOT_FOUND", "message": f"Task {task_id} not found"},
status_code=200
)
return JSONResponse(
content={
"task_id": result.task_id,
"status": result.status.value,
"error": result.error,
"created_at": result.created_at.isoformat() if result.created_at else None,
"started_at": result.started_at.isoformat() if result.started_at else None,
"completed_at": result.completed_at.isoformat() if result.completed_at else None,
},
status_code=200
)
except Exception as e:
logger.error(f"Error getting task status {task_id}: {e}")
raise APIException(ErrorCode.UNKNOWN_ERROR)
@app.post("/api/task/{task_id}/stop")
async def stop_task(task_id):
_validate_task_id(task_id)
logger.info("Start stopping ray job...")
success_json_info = JSONResponse(
content={"status": "Success", "message": f"{task_id} has been stopped"},
status_code=200
)
try:
executor_type = get_from_cfg(task_id, "executor_type")
if not WRAPPERS.get(executor_type).cancel(task_id):
raise APIException(ErrorCode.CANCEL_TASK_ERROR)
except Exception as e:
if isinstance(e, APIException):
raise e
raise APIException(ErrorCode.UNKNOWN_ERROR)
logger.info(f"{task_id} has been stopped.")
return success_json_info
def check_valid_path(file_path):
full_path = os.path.abspath(file_path)
return os.path.exists(full_path)
def get_from_cfg(task_id, key, default=None):
# Build path, then resolve ".." and verify it stays within /flow (CodeQL)
config_path = os.path.join("/flow", task_id, "process.yaml")
config_path = os.path.normpath(config_path)
if not config_path.startswith("/flow/"):
raise APIException(ErrorCode.PARAM_ERROR, detail=f"Invalid task_id: {task_id}")
if not check_valid_path(config_path):
logger.error(f"config_path is not existed! please check this path.")
raise APIException(ErrorCode.FILE_NOT_FOUND_ERROR)
with open(config_path, "r", encoding='utf-8') as f:
content = f.read()
cfg = yaml.safe_load(content)
return cfg.get(key, default)
def parse_args():
parser = ArgumentParser(description="Create API for Submitting Job to Data-juicer")
parser.add_argument(
'--ip',
type=str,
default="0.0.0.0",
help='Service ip for this API, default to use 0.0.0.0.'
)
parser.add_argument(
'--port',
type=int,
default=8080,
help='Service port for this API, default to use 8600.'
)
return parser.parse_args()
if __name__ == '__main__':
p_args = parse_args()
uvicorn.run(
app,
host=p_args.ip,
port=p_args.port
)
@@ -0,0 +1,25 @@
# -*- coding: utf-8 -*-
import importlib
import os
import re
import sys
from pathlib import Path
from loguru import logger
# 获取当前目录
current_dir = os.path.dirname(__file__)
# 遍历子目录
for module_name in os.listdir(current_dir):
module_path = os.path.join(current_dir, module_name)
# 检查是否是目录且包含 __init__.py,且 module_name 为合法的 Python 标识符
if (os.path.isdir(module_path)
and '__init__.py' in os.listdir(module_path)
and re.match(r'^[A-Za-z_][A-Za-z0-9_]*$', module_name)):
# 动态导入模块
try:
importlib.import_module(f".{module_name}", package=__name__)
except Exception as e:
logger.error(f"Failed to load Ops {module_name}")
@@ -0,0 +1,8 @@
from .cmd_task_scheduler import CommandScheduler
from .func_task_scheduler import CallableScheduler
from .job_task_scheduler import RayJobScheduler
cmd_scheduler = CommandScheduler(max_concurrent=5)
func_scheduler = CallableScheduler(max_concurrent=5)
ray_job_scheduler = RayJobScheduler(max_concurrent=5)
@@ -0,0 +1,229 @@
import asyncio
import os
from datetime import datetime
from typing import Optional, List
from loguru import logger
from .scheduler import Task, TaskStatus, TaskResult, TaskScheduler
class CommandTask(Task):
"""命令任务包装类"""
def __init__(self, task_id: str, command: str = None, log_path = None, shell: bool = True,
timeout: Optional[int] = None, cmd_args: List[str] = None, *args, **kwargs):
super().__init__(task_id, *args, **kwargs)
self.max_backups = 9
self.log_path = log_path
self.command = command
self.cmd_args = cmd_args
self.shell = shell if cmd_args is None else False
self.timeout = timeout
self.return_code = None
self._process = None
def start(self) -> 'CommandTask':
"""启动任务"""
if self.status == TaskStatus.PENDING:
self.status = TaskStatus.RUNNING
self.started_at = datetime.now()
self._task = asyncio.create_task(self._execute())
return self
async def _execute(self):
"""执行命令"""
try:
self.status = TaskStatus.RUNNING
self.started_at = datetime.now()
current_log_path = self.log_path
if os.path.exists(current_log_path):
counter = 1
while os.path.exists(f"{self.log_path}.{counter}"):
counter += 1
current_log_path = f"{self.log_path}.{counter}"
with open(current_log_path, 'a') as f:
if self.cmd_args is not None:
# Safe: use argument list with create_subprocess_exec (no shell injection)
process = await asyncio.create_subprocess_exec(
*self.cmd_args,
stdout=f,
stderr=asyncio.subprocess.STDOUT,
**self.kwargs
)
elif self.shell:
process = await asyncio.create_subprocess_shell(
self.command,
stdout=f,
stderr=asyncio.subprocess.STDOUT,
**self.kwargs
)
else:
process = await asyncio.create_subprocess_exec(
*self.command.split(),
stdout=f,
stderr=asyncio.subprocess.STDOUT,
**self.kwargs
)
self._process = process
# 等待进程完成(带超时)
try:
if self.timeout:
await asyncio.wait_for(
process.wait(),
timeout=self.timeout
)
else:
await process.wait()
self.return_code = process.returncode
if self._cancelled:
self.status = TaskStatus.CANCELLED
elif process.returncode == 0:
self.status = TaskStatus.COMPLETED
else:
self.status = TaskStatus.FAILED
except asyncio.TimeoutError:
# 超时处理
self._process.terminate()
try:
await asyncio.wait_for(self._process.wait(), timeout=5.0)
except asyncio.TimeoutError:
self._process.kill()
await self._process.wait()
self.status = TaskStatus.FAILED
f.write(f"\nCommand timed out after {self.timeout} seconds\n")
except asyncio.CancelledError:
# 任务被取消
if self._process:
self._process.terminate()
try:
await asyncio.wait_for(self._process.wait(), timeout=5.0)
except asyncio.TimeoutError:
self._process.kill()
await self._process.wait()
self.status = TaskStatus.CANCELLED
self._cancelled = True
except Exception as e:
self.status = TaskStatus.FAILED
logger.error(f"Task(id: {self.task_id}) run failed. Cause: {e}")
finally:
self.completed_at = datetime.now()
def cancel(self) -> bool:
"""取消任务"""
if self._process and self.status == TaskStatus.RUNNING:
try:
# 尝试优雅终止
self._process.terminate()
self._cancelled = True
return True
except Exception:
# 如果无法终止,强制杀死
try:
self._process.kill()
self._cancelled = True
return True
except Exception:
return False
return False
def to_result(self) -> TaskResult:
"""转换为结果对象"""
self.result = {
"command": self.command or self.cmd_args,
"return_code": self.return_code,
}
return super().to_result()
class CommandScheduler(TaskScheduler):
"""命令调度器"""
def __init__(self, max_concurrent: int = 5):
super().__init__(max_concurrent)
async def submit(self, task_id, command: str = None, log_path = None, shell: bool = True,
timeout: Optional[int] = None, cmd_args: List[str] = None, **kwargs) -> str:
if log_path is None:
log_path = f"/flow/{task_id}/output.log"
"""提交命令任务"""
task = CommandTask(task_id, command, log_path, shell, timeout, cmd_args=cmd_args, **kwargs)
self.tasks[task_id] = task
# 使用信号量限制并发
async with self.semaphore:
# 异步执行任务
task.start()
logger.info(f"命令任务 {task_id} 已提交并开始执行")
return task_id
def get_task_status(self, task_id: str) -> Optional[TaskResult]:
"""获取任务状态"""
task = self.tasks.get(task_id)
if task:
return task.to_result()
return None
def get_all_tasks(self) -> List[TaskResult]:
"""获取所有任务状态"""
return [task.to_result() for task in self.tasks.values()]
def cancel_task(self, task_id: str) -> bool:
"""取消任务"""
task = self.tasks.get(task_id)
if not task:
return True
if task.status == TaskStatus.RUNNING:
cancelled = task.cancel()
if cancelled:
logger.info(f"命令任务 {task_id} 已取消")
return cancelled
return False
def get_tasks_by_status(self, status: TaskStatus) -> List[TaskResult]:
"""根据状态获取任务"""
return [
task.to_result()
for task in self.tasks.values()
if task.status == status
]
async def wait_for_task(self, task_id: str, timeout: Optional[float] = None) -> TaskResult:
"""等待任务完成"""
task = self.tasks.get(task_id)
if not task:
raise ValueError(f"任务 {task_id} 不存在")
if task.status in [TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED]:
return task.to_result()
# 对于运行中的任务,我们已经通过 await task.execute() 等待了
# 所以这里直接返回结果
return task.to_result()
async def shutdown(self):
"""关闭调度器,取消所有运行中的任务"""
logger.info("正在关闭命令调度器...")
running_tasks = [
task for task in self.tasks.values()
if task.status == TaskStatus.RUNNING
]
for task in running_tasks:
logger.info(f"取消运行中的命令任务: {task.task_id}")
task.cancel()
logger.info("命令调度器已关闭")
@@ -0,0 +1,133 @@
import asyncio
from datetime import datetime
from typing import Callable, Optional, List
from loguru import logger
from .scheduler import TaskStatus, TaskResult, Task, TaskScheduler
class CallableTask(Task):
"""任务包装类"""
def __init__(self, task_id: str, func: Callable, *args, **kwargs):
super().__init__(task_id, *args, **kwargs)
self.func = func
def start(self) -> 'CallableTask':
"""启动任务"""
if self.status == TaskStatus.PENDING:
self.status = TaskStatus.RUNNING
self.started_at = datetime.now()
self._task = asyncio.create_task(self._execute())
return self
async def _execute(self):
"""执行任务"""
try:
self.result = await self.func(*self.args, **self.kwargs)
self.status = TaskStatus.COMPLETED
except asyncio.CancelledError:
self.status = TaskStatus.CANCELLED
self._cancelled = True
except Exception as e:
self.status = TaskStatus.FAILED
self.error = str(e)
finally:
self.completed_at = datetime.now()
def cancel(self) -> bool:
"""取消任务"""
if self._task and not self._task.done():
self._task.cancel()
return True
return False
class CallableScheduler(TaskScheduler):
"""异步任务调度器"""
def __init__(self, max_concurrent: int = 10):
super().__init__(max_concurrent)
async def submit(self, task_id, func: Callable, *args, **kwargs) -> str:
"""提交任务"""
task = CallableTask(task_id, func, *args, **kwargs)
self.tasks[task_id] = task
# 使用信号量限制并发
async with self.semaphore:
task.start()
logger.info(f"任务 {task_id} 已提交并开始执行")
return task_id
def get_task_status(self, task_id: str) -> Optional[TaskResult]:
"""获取任务状态"""
task = self.tasks.get(task_id)
if task:
return task.to_result()
return None
def get_all_tasks(self) -> List[TaskResult]:
"""获取所有任务状态"""
return [task.to_result() for task in self.tasks.values()]
def cancel_task(self, task_id: str) -> bool:
"""取消任务"""
task = self.tasks.get(task_id)
if task and task.status == TaskStatus.RUNNING:
cancelled = task.cancel()
if cancelled:
logger.info(f"任务 {task_id} 已取消")
return cancelled
return False
def get_tasks_by_status(self, status: TaskStatus) -> List[TaskResult]:
"""根据状态获取任务"""
return [
task.to_result()
for task in self.tasks.values()
if task.status == status
]
async def wait_for_task(self, task_id: str, timeout: Optional[float] = None) -> TaskResult:
"""等待任务完成"""
task = self.tasks.get(task_id)
if not task:
raise ValueError(f"任务 {task_id} 不存在")
if task.status in [TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED]:
return task.to_result()
# 等待任务完成
if task.get():
try:
await asyncio.wait_for(task.get(), timeout=timeout)
except asyncio.TimeoutError:
raise TimeoutError(f"任务 {task_id} 超时")
return task.to_result()
async def shutdown(self):
"""关闭调度器,取消所有运行中的任务"""
logger.info("正在关闭调度器...")
running_tasks = [
task for task in self.tasks.values()
if task.status == TaskStatus.RUNNING
]
for task in running_tasks:
logger.info(f"取消运行中的任务: {task.task_id}")
task.cancel()
# 等待所有任务完成
for task in running_tasks:
if task.get() and not task.get().done():
try:
await asyncio.wait_for(task.get(), timeout=5.0)
except asyncio.TimeoutError:
logger.warning(f"任务 {task.task_id} 无法正常停止")
logger.info("调度器已关闭")
@@ -0,0 +1,363 @@
import asyncio
import os
from datetime import datetime
from typing import Optional, List, Dict, Any
from loguru import logger
from .scheduler import Task, TaskStatus, TaskResult, TaskScheduler
# Default Ray dashboard address
RAY_DASHBOARD_ADDRESS = os.getenv("RAY_DASHBOARD_ADDRESS", "http://deepdata-raycluster-head-svc:8265")
class RayJobTask(Task):
"""Ray Job 任务包装类"""
def __init__(
self,
task_id: str,
entrypoint: str,
runtime_env: Optional[Dict[str, Any]] = None,
log_path: Optional[str] = None,
timeout: Optional[int] = None,
*args,
**kwargs,
):
super().__init__(task_id, *args, **kwargs)
self.entrypoint = entrypoint
self.runtime_env = runtime_env or {}
self.log_path = log_path or f"/flow/{task_id}/output.log"
self.timeout = timeout
self.job_id: Optional[str] = None
self._client = None
def _get_client(self):
"""延迟初始化 JobSubmissionClient"""
if self._client is None:
from ray.job_submission import JobSubmissionClient
self._client = JobSubmissionClient(RAY_DASHBOARD_ADDRESS)
return self._client
def start(self) -> "RayJobTask":
"""启动任务"""
if self.status == TaskStatus.PENDING:
self.status = TaskStatus.RUNNING
self.started_at = datetime.now()
self._task = asyncio.create_task(self._execute())
return self
async def _execute(self):
"""执行 Ray Job"""
try:
self.status = TaskStatus.RUNNING
self.started_at = datetime.now()
client = self._get_client()
# 确保 log 目录存在
log_dir = os.path.dirname(self.log_path)
if log_dir:
os.makedirs(log_dir, exist_ok=True)
# 提交 Ray Job
self.job_id = client.submit_job(
entrypoint=self.entrypoint,
runtime_env=self.runtime_env,
metadata={"task_id": self.task_id},
)
logger.info(f"Submitted Ray Job: {self.job_id} for task {self.task_id}")
# 轮询 Job 状态
poll_interval = 2 # 2 秒轮询一次
elapsed_time = 0
last_log_position = 0 # 记录已写入的日志位置
connection_failure_count = 0 # 连接失败计数器
max_connection_failures = 5 # 最大连接失败次数阈值
# 任务无进展超时:如果 job 一直是 RUNNING 但日志/进度没有任何变化,
# 说明 workers 可能已全部丢失,判定为失败
stall_timeout = int(os.getenv("RAY_JOB_STALL_TIMEOUT", "3600")) # 默认 120 秒
last_log_size = 0
last_active_time = datetime.now()
while True:
if self._cancelled:
logger.info(
f"Task {self.task_id} cancelled, stopping Ray Job {self.job_id}"
)
self._stop_job(client)
self.status = TaskStatus.CANCELLED
break
try:
info = client.get_job_info(self.job_id)
job_status = info.status
# 连接成功,重置失败计数器
connection_failure_count = 0
# 获取并写入日志
await self._fetch_and_write_logs(client, last_log_position)
current_log_size = os.path.getsize(self.log_path) if os.path.exists(self.log_path) else 0
last_log_position = current_log_size
# 检测任务是否停滞(workers 全部丢失但 head 仍返回 RUNNING)
if current_log_size > last_log_size:
last_active_time = datetime.now()
last_log_size = current_log_size
elif (datetime.now() - last_active_time).total_seconds() > stall_timeout:
logger.error(
f"Ray Job {self.job_id} stalled for {stall_timeout}s "
f"(no log progress), marking as FAILED"
)
self._stop_job(client)
self.status = TaskStatus.FAILED
self.error = f"Job stalled: no progress for {stall_timeout} seconds (workers may be lost)"
break
if job_status == "SUCCEEDED":
self.status = TaskStatus.COMPLETED
logger.info(f"Ray Job {self.job_id} completed successfully")
break
elif job_status == "FAILED":
self.status = TaskStatus.FAILED
self.error = info.message or "Job failed"
logger.error(f"Ray Job {self.job_id} failed: {self.error}")
break
elif job_status == "STOPPED":
if self._cancelled:
self.status = TaskStatus.CANCELLED
else:
self.status = TaskStatus.FAILED
self.error = "Job was stopped unexpectedly"
logger.info(f"Ray Job {self.job_id} stopped")
break
# 检查超时
if self.timeout and elapsed_time >= self.timeout:
logger.warning(
f"Ray Job {self.job_id} timed out after {self.timeout} seconds"
)
self._stop_job(client)
self.status = TaskStatus.FAILED
self.error = f"Job timed out after {self.timeout} seconds"
break
except (ConnectionError, ConnectionRefusedError, ConnectionResetError, OSError) as e:
# 连接类异常:Ray 集群可能已下线
connection_failure_count += 1
logger.error(
f"Connection to Ray cluster failed (attempt {connection_failure_count}/{max_connection_failures}): {e}"
)
if connection_failure_count >= max_connection_failures:
self.status = TaskStatus.FAILED
self.error = f"Lost connection to Ray cluster after {max_connection_failures} attempts: {str(e)}"
logger.error(f"Task {self.task_id} failed: {self.error}")
break
except Exception as e:
# 其他异常:记录警告但继续重试
logger.warning(f"Error checking job status: {e}")
await asyncio.sleep(poll_interval)
elapsed_time += poll_interval
except asyncio.CancelledError:
logger.info(f"Task {self.task_id} received CancelledError")
if self.job_id:
client = self._get_client()
self._stop_job(client)
self.status = TaskStatus.CANCELLED
self._cancelled = True
except Exception as e:
self.status = TaskStatus.FAILED
self.error = str(e)
logger.error(f"RayJobTask(id: {self.task_id}) run failed. Cause: {e}")
finally:
self.completed_at = datetime.now()
async def _fetch_and_write_logs(self, client, last_position: int = 0):
"""获取 Ray Job 日志并追加写入日志文件"""
try:
logs = client.get_job_logs(self.job_id)
if logs:
log_content = logs if isinstance(logs, str) else str(logs)
# 只追加新日志
if len(log_content) > last_position:
new_logs = log_content[last_position:]
with open(self.log_path, "a", encoding="utf-8") as f:
f.write(new_logs)
if not new_logs.endswith("\n"):
f.write("\n")
except Exception as e:
logger.warning(f"Failed to fetch logs for job {self.job_id}: {e}")
def _stop_job(self, client):
"""停止 Ray Job"""
if self.job_id:
try:
client.stop_job(self.job_id)
logger.info(f"Stopped Ray Job {self.job_id}")
except Exception as e:
logger.warning(f"Failed to stop Ray Job {self.job_id}: {e}")
def cancel(self) -> bool:
"""取消任务"""
if self.status == TaskStatus.RUNNING:
self._cancelled = True
logger.info(f"Marked Ray Job task {self.task_id} for cancellation")
return True
return False
def to_result(self) -> TaskResult:
"""转换为结果对象"""
self.result = {
"entrypoint": self.entrypoint,
"job_id": self.job_id,
}
return super().to_result()
class RayJobScheduler(TaskScheduler):
"""Ray Job 调度器"""
def __init__(self, max_concurrent: int = 5, ray_address: Optional[str] = None):
super().__init__(max_concurrent)
self.ray_address = ray_address or RAY_DASHBOARD_ADDRESS
self._client = None
def _get_client(self):
"""延迟初始化 JobSubmissionClient"""
if self._client is None:
from ray.job_submission import JobSubmissionClient
self._client = JobSubmissionClient(self.ray_address)
return self._client
async def submit(
self,
task_id: str,
script_path: str,
script_args: str = "",
runtime_env: Optional[Dict[str, Any]] = None,
log_path: Optional[str] = None,
timeout: Optional[int] = None,
**kwargs,
) -> str:
"""提交 Ray Job 任务"""
# 构建 entrypoint
entrypoint = f"python {script_path}"
if script_args:
entrypoint = f"{entrypoint} {script_args}"
# 默认 runtime_env
if runtime_env is None:
runtime_env = {"working_dir": "/opt/runtime"}
# 默认 log_path
if log_path is None:
log_path = f"/flow/{task_id}/output.log"
task = RayJobTask(
task_id=task_id,
entrypoint=entrypoint,
runtime_env=runtime_env,
log_path=log_path,
timeout=timeout,
**kwargs,
)
self.tasks[task_id] = task
# 使用信号量限制并发
async with self.semaphore:
task.start()
logger.info(f"Ray Job 任务 {task_id} 已提交并开始执行")
return task_id
def get_task_status(self, task_id: str) -> Optional[TaskResult]:
"""获取任务状态"""
task = self.tasks.get(task_id)
if task:
return task.to_result()
return None
def get_all_tasks(self) -> List[TaskResult]:
"""获取所有任务状态"""
return [task.to_result() for task in self.tasks.values()]
def cancel_task(self, task_id: str) -> bool:
"""取消任务"""
task = self.tasks.get(task_id)
if not task:
logger.warning(f"Task {task_id} not found, considering already cancelled")
return True
if task.status == TaskStatus.RUNNING:
cancelled = task.cancel()
if cancelled:
logger.info(f"Ray Job 任务 {task_id} 已标记为取消")
return cancelled
# 任务未运行,直接返回成功
if task.status in [TaskStatus.PENDING, TaskStatus.COMPLETED, TaskStatus.FAILED]:
return True
return False
def get_tasks_by_status(self, status: TaskStatus) -> List[TaskResult]:
"""根据状态获取任务"""
return [
task.to_result() for task in self.tasks.values() if task.status == status
]
async def wait_for_task(
self, task_id: str, timeout: Optional[float] = None
) -> TaskResult:
"""等待任务完成"""
task = self.tasks.get(task_id)
if not task:
raise ValueError(f"任务 {task_id} 不存在")
if task.status in [
TaskStatus.COMPLETED,
TaskStatus.FAILED,
TaskStatus.CANCELLED,
]:
return task.to_result()
# 等待任务完成
if task.get():
try:
await asyncio.wait_for(task.get(), timeout=timeout)
except asyncio.TimeoutError:
raise TimeoutError(f"任务 {task_id} 等待超时")
return task.to_result()
async def shutdown(self):
"""关闭调度器,取消所有运行中的任务"""
logger.info("正在关闭 Ray Job 调度器...")
running_tasks = [
task for task in self.tasks.values() if task.status == TaskStatus.RUNNING
]
for task in running_tasks:
logger.info(f"取消运行中的 Ray Job 任务: {task.task_id}")
task.cancel()
# 等待所有任务完成
for task in running_tasks:
if task.get() and not task.get().done():
try:
await asyncio.wait_for(task.get(), timeout=5.0)
except asyncio.TimeoutError:
logger.warning(f"任务 {task.task_id} 无法正常停止")
logger.info("Ray Job 调度器已关闭")
@@ -0,0 +1,160 @@
# 任务状态枚举
import asyncio
import signal
import sys
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Any, Optional, Dict, List
from loguru import logger
class TaskStatus(Enum):
PENDING = "pending" # 等待执行
RUNNING = "running" # 正在运行
COMPLETED = "completed" # 已完成
FAILED = "failed" # 执行失败
CANCELLED = "cancelled" # 已取消
@dataclass
class TaskResult:
"""任务结果数据类"""
task_id: str
status: TaskStatus
result: Any = None
error: Optional[str] = None
created_at: datetime = None
started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None
progress: float = 0.0
class Task:
def __init__(self, task_id: str, *args, **kwargs):
self.task_id = task_id
self.args = args
self.kwargs = kwargs
self.status = TaskStatus.PENDING
self.result = None
self.error = None
self.created_at = datetime.now()
self.started_at = None
self.completed_at = None
self.progress = 0.0
self._task = None # asyncio.Task 实例
self._cancelled = False
def get(self):
return self._task
def start(self) -> 'Task':
"""启动任务"""
pass
async def _execute(self):
"""执行任务"""
pass
def cancel(self) -> bool:
"""取消任务"""
pass
def to_result(self) -> TaskResult:
"""转换为结果对象"""
return TaskResult(
task_id=self.task_id,
status=self.status,
result=self.result,
error=self.error,
created_at=self.created_at,
started_at=self.started_at,
completed_at=self.completed_at,
progress=self.progress
)
class TaskScheduler:
"""异步任务调度器"""
def __init__(self, max_concurrent: int = 10):
self.max_concurrent = max_concurrent
self.tasks: Dict[str, Task] = {}
self.semaphore = asyncio.Semaphore(max_concurrent)
# 注册信号处理器
try:
signal.signal(signal.SIGINT, self._signal_handler)
signal.signal(signal.SIGTERM, self._signal_handler)
except (ValueError, AttributeError):
# 在某些平台上可能不支持
pass
def _signal_handler(self, signum, frame):
"""信号处理器"""
logger.info(f"收到信号 {signum},正在清理任务...")
asyncio.create_task(self.shutdown())
sys.exit(0)
async def submit(self, task_id, task, *args, **kwargs) -> str:
"""提交任务"""
pass
def get_task_status(self, task_id: str) -> Optional[TaskResult]:
"""获取任务状态"""
task = self.tasks.get(task_id)
if task:
return task.to_result()
return None
def get_all_tasks(self) -> List[TaskResult]:
"""获取所有任务状态"""
return [task.to_result() for task in self.tasks.values()]
def cancel_task(self, task_id: str) -> bool:
"""取消任务"""
task = self.tasks.get(task_id)
if task and task.status == TaskStatus.RUNNING:
cancelled = task.cancel()
if cancelled:
logger.info(f"任务 {task_id} 已取消")
return cancelled
return False
def get_tasks_by_status(self, status: TaskStatus) -> List[TaskResult]:
"""根据状态获取任务"""
return [
task.to_result()
for task in self.tasks.values()
if task.status == status
]
async def wait_for_task(self, task_id: str, timeout: Optional[float] = None) -> TaskResult:
"""等待任务完成"""
pass
async def shutdown(self):
"""关闭调度器,取消所有运行中的任务"""
pass
def get_statistics(self) -> Dict[str, int]:
"""获取统计信息"""
stats = {
TaskStatus.PENDING: 0,
TaskStatus.RUNNING: 0,
TaskStatus.COMPLETED: 0,
TaskStatus.FAILED: 0,
TaskStatus.CANCELLED: 0
}
for task in self.tasks.values():
stats[task.status] += 1
return {
"pending": stats[TaskStatus.PENDING],
"running": stats[TaskStatus.RUNNING],
"completed": stats[TaskStatus.COMPLETED],
"failed": stats[TaskStatus.FAILED],
"cancelled": stats[TaskStatus.CANCELLED],
"total": len(self.tasks)
}
@@ -0,0 +1,2 @@
@@ -0,0 +1,260 @@
# -*- coding: utf-8 -*-
import json
import os
import time
import uuid
from datetime import datetime
from pathlib import Path
from typing import Dict, Any
from loguru import logger
from sqlalchemy import text
from deepdata.sql_manager.sql_manager import SQLManager
class TaskInfoPersistence:
def __init__(self):
self.sql_dict = self.load_sql_dict()
@staticmethod
def load_sql_dict():
"""获取sql语句"""
sql_config_path = "/opt/runtime/deepdata/sql_manager/sql/sql_config.json"
with open(sql_config_path, 'r', encoding='utf-8') as f:
return json.load(f)
def update_task_result(self, sample, file_id = None):
if file_id is None:
file_id = str(uuid.uuid4())
instance_id = str(sample.get("instance_id"))
src_file_name = str(sample.get("sourceFileName"))
src_file_type = str(sample.get("sourceFileType"))
src_file_id = str(sample.get("sourceFileId"))
src_file_size = int(sample.get("sourceFileSize"))
file_size = str(sample.get("fileSize"))
file_type = str(sample.get("fileType"))
file_name = str(sample.get("fileName"))
status = str(sample.get("execute_status"))
failed_reason = json.dumps(sample.get("failed_reason"), ensure_ascii=False)
result_data = {
"instance_id": instance_id,
"src_file_id": src_file_id,
"dest_file_id": file_id,
"src_name": src_file_name,
"dest_name": file_name,
"src_type": src_file_type,
"dest_type": file_type,
"src_size": src_file_size,
"dest_size": file_size,
"status": status,
"result": failed_reason
}
self.insert_result(result_data, str(self.sql_dict.get("insert_clean_result_sql")))
def update_file_result(self, sample, file_id):
file_size = str(sample.get("fileSize"))
file_type = str(sample.get("fileType"))
file_name = str(sample.get("fileName"))
dataset_id = str(sample.get("dataset_id"))
file_path = str(sample.get("filePath"))
create_time = datetime.now()
last_access_time = datetime.fromtimestamp(os.path.getmtime(file_path))
file_data = {
"id": file_id,
"dataset_id": dataset_id,
"file_name": file_name,
"file_path": file_path,
"file_type": file_type,
"file_size": file_size,
"status": "COMPLETED",
"upload_time": create_time,
"last_access_time": last_access_time,
"created_at": create_time,
"updated_at": create_time
}
self.insert_result(file_data, str(self.sql_dict.get("insert_dataset_file_sql")))
def query_existing_files(self, dataset_id: str):
result = None
query_sql = str(self.sql_dict.get("query_dataset_files_sql"))
with SQLManager.create_connect() as conn:
execute_result = conn.execute(text(query_sql), {"dataset_id": dataset_id})
result = execute_result.fetchall()
return result
def batch_insert_files(self, samples):
insert_sql = str(self.sql_dict.get("insert_dataset_file_sql"))
self.batch_execute(insert_sql, samples)
def persistence_task_info(self, sample: Dict[str, Any]):
file_id = str(uuid.uuid4())
self.update_task_result(sample, file_id)
self.update_file_result(sample, file_id)
@staticmethod
def insert_result(data, sql):
retries = 0
max_retries = 20
retry_delay = 1
while retries <= max_retries:
try:
with SQLManager.create_connect() as conn:
conn.execute(text(sql), data)
return
except Exception as e:
if "database is locked" in str(e) or "locking protocol" in str(e):
retries += 1
time.sleep(retry_delay)
else:
logger.error("database execute failed: {}", str(e))
raise RuntimeError(82000, str(e)) from None
raise Exception("Max retries exceeded")
@staticmethod
def batch_execute(sql, args_list):
"""
批量执行 SQL
:param sql: SQL 语句 (例如: "INSERT INTO table (a, b) VALUES (%s, %s)")
:param args_list: 参数列表 (例如: [(1, 2), (3, 4), ...])
"""
# 获取连接
with SQLManager.create_connect() as conn:
try:
conn.execute(text(sql), args_list)
except Exception as e:
conn.rollback()
logger.error(f"批量插入失败: {e}")
raise e
def update_result(self, dataset_id, instance_id, status):
dataset_data = {
"dataset_id": dataset_id
}
query_dataset_sql = str(self.sql_dict.get("query_dataset_sql"))
with SQLManager.create_connect() as conn:
result = conn.execute(text(query_dataset_sql), dataset_data)
if result:
rows = result.fetchall()
total_size = sum(int(row[0]) for row in rows)
file_count = len(rows)
else:
total_size = 0
file_count = 0
dataset_data.update({
"task_id": instance_id,
"total_size": total_size,
"file_count": file_count
})
update_dataset_sql = str(self.sql_dict.get("update_dataset_sql"))
self.insert_result(dataset_data, update_dataset_sql)
# Determine final status based on file results
final_status = self._determine_task_status(instance_id, status)
task_data = {
"task_id": instance_id,
"status": final_status,
"total_size": total_size,
"finished_time": datetime.now()
}
update_task_sql = str(self.sql_dict.get("update_task_sql"))
self.insert_result(task_data, update_task_sql)
def _determine_task_status(self, instance_id: str, executor_status: str) -> str:
"""Determine final task status based on file processing results"""
if executor_status == "FAILED":
return "FAILED"
# Query file results to determine status
try:
with SQLManager.create_connect() as conn:
# Count completed and failed files from t_clean_result (same table as progress calculation)
query = text("""
SELECT
COUNT(CASE WHEN status = 'COMPLETED' THEN 1 END) as completed,
COUNT(CASE WHEN status = 'FAILED' THEN 1 END) as failed
FROM t_clean_result
WHERE instance_id = :instance_id
""")
result = conn.execute(query, {"instance_id": instance_id})
row = result.fetchone()
if row:
completed = row[0] or 0
failed = row[1] or 0
logger.info(f"Task {instance_id} file results: completed={completed}, failed={failed}")
# Determine status based on file results
if completed > 0 and failed > 0:
final_status = "PARTIAL_SUCCESS"
elif completed == 0 and failed == 0:
final_status = "COMPLETED"
elif completed > 0:
final_status = "COMPLETED"
else:
final_status = "FAILED"
logger.info(f"Task {instance_id} final status: {final_status}")
return final_status
except Exception as e:
logger.warning(f"Failed to determine task status for {instance_id}: {e}")
return executor_status
def query_task_info(self, instance_ids: list[str]):
result = {}
current_result = None
for instance_id in instance_ids:
try:
current_result = self.execute_sql_query(instance_id)
except Exception as e:
logger.warning("instance_id: {}, query job result error: {}", instance_id, str(e))
if current_result:
result[instance_id] = current_result
return result
def execute_sql_query(self, instance_id):
result = None
create_tables_sql = str(self.sql_dict.get("create_tables_sql"))
query_sql = str(self.sql_dict.get("query_sql"))
with SQLManager.create_connect() as conn:
conn.execute(text(create_tables_sql))
execute_result = conn.execute(text(query_sql), {"instance_id": instance_id})
result = execute_result.fetchall()
return result
# todo 删除接口待实现
def delete_task_info(self, instance_id: str):
create_tables_sql = self.sql_dict.get("create_tables_sql")
delete_task_instance_sql = self.sql_dict.get("delete_task_instance_sql")
try:
with SQLManager.create_connect() as conn:
conn.execute(text(create_tables_sql))
conn.execute(text(delete_task_instance_sql), {"instance_id": instance_id})
except Exception as e:
logger.warning(f"delete database for flow: {instance_id}", e)
def delete_task_operate_info(self, instance_id: str):
create_duplicate_img_tables_sql = self.sql_dict.get("create_duplicate_img_tables_sql")
create_similar_img_tables_sql = self.sql_dict.get("create_similar_img_tables_sql")
create_similar_text_tables_sql = self.sql_dict.get("create_similar_text_tables_sql")
delete_duplicate_img_tables_sql = self.sql_dict.get("delete_duplicate_img_tables_sql")
delete_similar_img_tables_sql = self.sql_dict.get("delete_similar_img_tables_sql")
delete_similar_text_tables_sql = self.sql_dict.get("delete_similar_text_tables_sql")
try:
with SQLManager.create_connect() as conn:
conn.execute(text(create_duplicate_img_tables_sql))
conn.execute(text(delete_duplicate_img_tables_sql), {"instance_id": instance_id})
conn.execute(text(create_similar_img_tables_sql))
conn.execute(text(delete_similar_img_tables_sql), {"instance_id": instance_id})
conn.execute(text(create_similar_text_tables_sql))
conn.execute(text(delete_similar_text_tables_sql), {"instance_id": instance_id})
except Exception as e:
logger.warning(f"delete database for flow: {instance_id} error", e)
@@ -0,0 +1,18 @@
{
"query_sql": "SELECT * FROM t_task_instance_info WHERE instance_id IN (:instance_id)",
"insert_sql": "INSERT INTO t_task_instance_info (instance_id, meta_file_name, meta_file_type, meta_file_id, meta_file_size, file_id, file_size, file_type, file_name, file_path, status, operator_id, error_code, incremental, child_id, slice_num) VALUES (:instance_id, :meta_file_name, :meta_file_type, :meta_file_id, :meta_file_size, :file_id, :file_size, :file_type, :file_name, :file_path, :status, :operator_id, :error_code, :incremental, :child_id, :slice_num)",
"insert_dataset_file_sql": "INSERT INTO t_dm_dataset_files (id, dataset_id, file_name, file_path, file_type, file_size, status, upload_time, last_access_time, created_at, updated_at) VALUES (:id, :dataset_id, :file_name, :file_path, :file_type, :file_size, :status, :upload_time, :last_access_time, :created_at, :updated_at)",
"insert_clean_result_sql": "INSERT INTO t_clean_result (instance_id, src_file_id, dest_file_id, src_name, dest_name, src_type, dest_type, src_size, dest_size, status, result) VALUES (:instance_id, :src_file_id, :dest_file_id, :src_name, :dest_name, :src_type, :dest_type, :src_size, :dest_size, :status, :result)",
"query_dataset_sql": "SELECT file_size FROM t_dm_dataset_files WHERE dataset_id = :dataset_id",
"update_dataset_sql": "UPDATE t_dm_datasets SET size_bytes = :total_size, file_count = :file_count WHERE id = :dataset_id;",
"update_task_sql": "UPDATE t_clean_task SET status = :status, after_size = :total_size, finished_at = :finished_time WHERE id = :task_id",
"create_tables_sql": "CREATE TABLE IF NOT EXISTS t_task_instance_info (instance_id VARCHAR(255), meta_file_name TEXT, meta_file_type VARCHAR(100), meta_file_id BIGINT, meta_file_size VARCHAR(100), file_id BIGINT, file_size VARCHAR(100), file_type VARCHAR(100), file_name TEXT, file_path TEXT, status INT, operator_id VARCHAR(255), error_code VARCHAR(100), incremental VARCHAR(50), child_id BIGINT, slice_num INT DEFAULT 0);",
"delete_task_instance_sql": "DELETE FROM t_task_instance_info WHERE instance_id = :instance_id",
"create_duplicate_img_tables_sql": "CREATE TABLE IF NOT EXISTS operator_duplicate_img_features (id SERIAL PRIMARY KEY, task_uuid VARCHAR(255), file_feature TEXT, file_name TEXT, timestamp TIMESTAMP);",
"delete_duplicate_img_tables_sql": "DELETE FROM operator_duplicate_img_features WHERE flow_id = :flow_id",
"create_similar_img_tables_sql": "CREATE TABLE IF NOT EXISTS operator_similar_img_features (id SERIAL PRIMARY KEY, task_uuid VARCHAR(255), p_hash TEXT, des_matrix BYTEA, matrix_shape TEXT, file_name TEXT, timestamp TIMESTAMP);",
"delete_similar_img_tables_sql": "DELETE FROM operator_similar_img_features WHERE flow_id = :flow_id",
"create_similar_text_tables_sql": "CREATE TABLE IF NOT EXISTS operators_similar_text_features (id SERIAL PRIMARY KEY, task_uuid VARCHAR(255), file_feature TEXT, file_name TEXT, timestamp TIMESTAMP);",
"delete_similar_text_tables_sql": "DELETE FROM operators_similar_text_features WHERE flow_id = :flow_id",
"query_dataset_files_sql": "SELECT file_path FROM t_dm_dataset_files WHERE dataset_id = :dataset_id"
}
@@ -0,0 +1,83 @@
# -- encoding: utf-8 --
import os
import time
from random import uniform
from threading import Lock
from loguru import logger
from sqlalchemy import create_engine
from sqlalchemy.engine import URL
from urllib.parse import quote_plus
class SQLManager:
_engine = None
_lock = Lock() # 确保多线程环境下只创建一个引擎
@classmethod
def _get_engine(cls):
"""
单例模式获取 Engine,确保全局只有一个连接池
"""
if cls._engine is not None:
return cls._engine
with cls._lock:
if cls._engine is not None:
return cls._engine
db_user = os.getenv("PG_USER", "postgres")
db_pass = os.getenv("PG_PASSWORD", "password")
db_host = os.getenv("PG_HOST", "deepdata-database")
db_port = int(os.getenv("PG_PORT", 5432))
db_name = os.getenv("PG_DATABASE", "deepdata")
# 🔍 强制打印日志,确认环境变量是否真正注入
logger.info(f"🔑 DB Config from ENV -> Host:{db_host}, User:{db_user}, Pass:{db_pass}")
# ✅ URL 编码:防止密码中的 # @ / 等特殊字符破坏连接字符串
encoded_pass = quote_plus(db_pass)
connection_url = f"postgresql+psycopg2://{db_user}:{encoded_pass}@{db_host}:{db_port}/{db_name}"
logger.info(f"🔗 Connecting to: {connection_url.replace(encoded_pass, '***')}")
# 创建引擎 (只执行一次)
# 注意:AUTOCOMMIT 虽然方便,但建议根据业务场景谨慎使用。
# 如果需要事务控制(比如两张表必须同时更新成功),AUTOCOMMIT 会导致无法回滚。
cls._engine = create_engine(
connection_url,
pool_pre_ping=True,
isolation_level="AUTOCOMMIT",
pool_size=5, # 显式指定池大小
max_overflow=15, # 显式指定溢出
pool_timeout=30,
pool_recycle=1800 # 10分钟回收连接
)
logger.info("Database Engine initialized successfully.")
return cls._engine
@staticmethod
def create_connect(max_retries=5, base_delay=1):
"""
从现有的 Engine 连接池中获取连接,包含重试逻辑
"""
attempt = 0
while True:
try:
# 1. 获取全局引擎
engine = SQLManager._get_engine()
# 2. 从池中借出一个连接
return engine.connect()
except Exception as e:
attempt += 1
logger.error(f"Connection attempt {attempt} failed: {str(e)}")
if attempt >= max_retries:
logger.error("Max retries reached. Could not connect to database.")
raise
# 重试等待逻辑
wait_time = min(30, base_delay * (2 ** (attempt - 1)))
jitter = uniform(-wait_time / 4, wait_time / 4)
sleep_time = wait_time + jitter
logger.info(f"Retrying in {sleep_time:.2f} seconds...")
time.sleep(sleep_time)
@@ -0,0 +1,7 @@
from . import data_juicer_wrapper, deepdata_wrapper
WRAPPERS = {
"ray": data_juicer_wrapper,
"default": data_juicer_wrapper,
"deepdata": deepdata_wrapper
}
@@ -0,0 +1,176 @@
import base64
import time
from json import dumps as jdumps
from json import loads as jloads
from typing import Dict, Optional
from urllib.parse import urljoin
import ray
import requests
import yaml
from jsonargparse import ArgumentParser
from loguru import logger
from deepdata.core.base_op import FileExporter, SUCCESS_STATUS
from deepdata.core.constant import Fields
from deepdata.wrappers.executor import RayExecutor
from deepdata.sql_manager.persistence_atction import TaskInfoPersistence
DJ_OUTPUT = "outputs"
class DataJuicerClient:
def __init__(self, base_url):
self.base_url = base_url
def call_data_juicer_api(self, path: str, params: Optional[Dict] = None, json: Optional[Dict] = None):
url = urljoin(self.base_url, path)
if json is not None:
response = requests.post(url, params=params, json=json)
else:
response = requests.get(url, params=params)
return jloads(response.text)
def init_config(self, dataset_path: str, export_path, process):
"""
Initialize Data-Juicer config.
Args:
:param dataset_path: The input dataset path.
:param process: The ops
:param export_path: The export path.
"""
dj_config = {
"dataset_path": dataset_path,
"export_path": export_path,
"process": process,
"executor_type": "default",
}
url_path = "/data_juicer/config/get_init_configs"
try:
res = self.call_data_juicer_api(url_path, params={"cfg": jdumps(dj_config)})
except Exception as e:
error_msg = f"An unexpected error occurred in calling {url_path}:\n{e}"
raise RuntimeError(error_msg)
return res["result"]
def execute_config(self, dj_config: Dict):
"""
Execute data-juicer data process.
Args:
dj_config: configs of data-juicer
"""
url_path = "/data_juicer/core/DefaultExecutor/run"
try:
res = self.call_data_juicer_api(url_path, params={"skip_return": True}, json={"cfg": jdumps(dj_config)})
if res.get("status") != "success":
raise RuntimeError(f"An error occurred in calling {url_path}:\n{res}")
return dj_config["export_path"]
except Exception as e:
error_msg = f"An unexpected error occurred in calling {url_path}:\n{e}"
raise RuntimeError(error_msg)
class DataJuicerExecutor(RayExecutor):
def __init__(self, cfg = None, meta = None):
super().__init__(cfg, meta)
self.client = DataJuicerClient(base_url="http://deepdata-data-juicer:8000")
self.dataset_path = f"/flow/{self.cfg.instance_id}/dataset_on_dj.jsonl"
self.export_path = f"/flow/{self.cfg.instance_id}/processed_dataset.jsonl"
def add_column(self, batch):
batch_size = len(batch["filePath"])
batch["execute_status"] = [SUCCESS_STATUS] * batch_size
batch["executor"] = ["datajuicer"] * batch_size
batch[Fields.instance_id] = [self.cfg.instance_id] * batch_size
batch[Fields.export_path] = [self.cfg.export_path] * batch_size
return batch
def run(self):
# 1. 加载数据集
logger.info('Loading dataset with Ray...')
if self.meta:
file_content = base64.b64decode(self.meta)
lines = file_content.splitlines()
dataset = ray.data.from_items([jloads(line) for line in lines])
else:
dataset = self.load_dataset()
logger.info('Read data...')
dataset = dataset.map(FileExporter().convert_to_dj, num_cpus=0.05)
# 保存原始数据文件ID集合,用于后续过滤数据检测
original_file_ids = set(dataset.unique("fileId"))
# 写入数据集文件
with open(self.dataset_path, "w", encoding="utf-8") as f:
for batch_df in dataset.iter_batches(batch_format="pandas", batch_size=2048):
batch_df.to_json(f, orient="records", lines=True, force_ascii=False)
logger.info('Processing data...')
tstart = time.time()
try:
dj_config = self.client.init_config(self.dataset_path, self.export_path, self.cfg.process)
result_path = self.client.execute_config(dj_config)
processed_dataset = self.load_dj_dataset(result_path)
processed_dataset = processed_dataset.map_batches(self.add_column, num_cpus=0.05)
processed_dataset = processed_dataset.map(FileExporter().save_file_and_db, num_cpus=0.05)
processed_dataset = processed_dataset.materialize()
# 特殊处理:识别被过滤的数据
if processed_dataset.count() == 0:
processed_file_ids = set()
else:
processed_file_ids = set(processed_dataset.unique("fileId"))
filtered_file_ids = original_file_ids - processed_file_ids
if filtered_file_ids:
logger.info(f"Found {len(filtered_file_ids)} filtered files, updating task result only")
for sample_dict in dataset.iter_batches(batch_format="pandas", batch_size=2048):
for _, row in sample_dict.iterrows():
if str(row.get("fileId", "")) in filtered_file_ids:
row["fileSize"] = "0"
row["fileType"] = ""
row["execute_status"] = SUCCESS_STATUS
row[Fields.instance_id] = self.cfg.instance_id
TaskInfoPersistence().update_task_result(row)
self.scan_files()
except Exception as e:
logger.error(f"An unexpected error occurred.", e)
raise e
tend = time.time()
logger.info(f'All Ops are done in {tend - tstart:.3f}s.')
if __name__ == '__main__':
parser = ArgumentParser(description="Create API for Submitting Job to Data-juicer")
parser.add_argument("--config_path", type=str, required=False, default="../configs/demo.yaml")
parser.add_argument("--flow_config", type=str, required=False, default=None)
args = parser.parse_args()
config_path = args.config_path
flow_config = args.flow_config
if flow_config:
m_cfg = yaml.safe_load(base64.b64decode(flow_config))
else:
with open(config_path, "r", encoding='utf-8') as f:
m_cfg = yaml.safe_load(f)
executor = DataJuicerExecutor(m_cfg)
try:
executor.run()
except Exception as e:
executor.update_db("FAILED")
raise e
executor.update_db("COMPLETED")
@@ -0,0 +1,18 @@
# -*- coding: utf-8 -*-
import os
from deepdata.scheduler import cmd_scheduler
async def submit(task_id, config_path):
current_dir = os.path.dirname(__file__)
executor_script = os.path.join(current_dir, 'data_juicer_executor.py')
# Use argument list to avoid shell injection (CodeQL / FCE)
await cmd_scheduler.submit(
task_id,
cmd_args=["python", executor_script, f"--config_path={config_path}"]
)
def cancel(task_id):
return cmd_scheduler.cancel_task(task_id)
@@ -0,0 +1,152 @@
# -*- coding: utf-8 -*-
import base64
import json
import os
import subprocess
import time
import ray
import yaml
from jsonargparse import ArgumentParser
from loguru import logger
from deepdata.core.dataset import RayDataset
from deepdata.wrappers.executor import RayExecutor
import deepdata.ops
# xuhuanqing
class DeepDataExecutor(RayExecutor):
"""
基于Ray的统一执行器.
1. 支持原生 Ray Dataset 算子任务 (Mapper, Filter 等)
2. 支持外部 CLI 任务 (如 DataX, Shell 命令等)
"""
def __init__(self, cfg=None, meta=None):
super().__init__(cfg, meta)
# 提取任务实例ID,用于日志路径和文件注册
self.instance_id = getattr(self.cfg, 'instance_id', 'unknown')
def run(self):
# 🔄 路由分发:检查配置中是否包含 CLI 任务
cli_cfg = getattr(self.cfg, 'cli_config', None)
if cli_cfg and cli_cfg.get('command'):
logger.info("📦 Routing to CLI execution mode...")
self._run_cli_task(cli_cfg)
else:
logger.info("🔄 Routing to native Ray Dataset execution mode...")
self._run_ray_dataset()
def _run_cli_task(self, cli_cfg):
"""
执行外部命令行任务 (如 DataX)
特点:同步阻塞、日志实时重定向、支持超时控制
"""
cmd = [cli_cfg['command']] + cli_cfg.get('args', [])
timeout = cli_cfg.get('timeout')
# 准备日志目录
log_path = getattr(self.cfg, 'log_path', None)
if log_path == None:
log_path = f"/flow/{self.instance_id}/output.log"
os.makedirs(os.path.dirname(log_path), exist_ok=True)
logger.info(f"🚀 Executing CLI command: {' '.join(cmd)}")
logger.info(f"📝 Logs will be redirected to: {log_path}")
try:
with open(log_path, "w", encoding="utf-8") as log_file:
# 启动子进程,合并 stdout 和 stderr
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1,
encoding='utf-8'
)
# 流式读取并写入日志文件(同时同步到 Ray Dashboard)
for line in process.stdout:
line = line.rstrip()
if line:
log_file.write(line + "\n")
log_file.flush()
logger.info(line)
# 等待进程完成(支持超时)
process.wait(timeout=timeout)
if process.returncode != 0:
raise RuntimeError(f"CLI task failed with exit code {process.returncode}")
logger.info("✅ CLI task completed successfully.")
# CLI 任务完成后,扫描可能的输出文件并注册到元数据
# self.scan_files()
except subprocess.TimeoutExpired:
process.kill()
logger.error(f"⏱️ CLI task timed out after {timeout} seconds")
raise RuntimeError(f"CLI task timed out after {timeout} seconds")
except Exception as e:
logger.error(f"❌ CLI task execution failed: {e}")
raise
def _run_ray_dataset(self):
"""
执行原生 Ray Dataset 算子任务(保持原有逻辑)
"""
logger.info('Loading dataset with Ray...')
if self.meta:
file_content = base64.b64decode(self.meta)
lines = file_content.splitlines()
dataset = ray.data.from_items([json.loads(line) for line in lines])
else:
dataset = self.load_dataset()
dataset = RayDataset(dataset, self.cfg)
logger.info('Processing data...')
tstart = time.time()
dataset.process(self.cfg.process, **getattr(self.cfg, 'kwargs', {}))
tend = time.time()
logger.info(f'All Ops are done in {tend - tstart:.3f}s.')
dataset.data.materialize()
self.scan_files()
if __name__ == '__main__':
parser = ArgumentParser(description="Unified Executor for Ray & CLI Tasks")
parser.add_argument("--config_path", type=str, required=False, default="../configs/demo.yaml")
parser.add_argument("--flow_config", type=str, required=False, default=None)
args = parser.parse_args()
logger.info("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
logger.info(f"{args}")
config_path = args.config_path
flow_config = args.flow_config
if flow_config:
m_cfg = yaml.safe_load(base64.b64decode(flow_config))
else:
with open(config_path, "r", encoding='utf-8') as f:
m_cfg = yaml.safe_load(f)
executor = DeepDataExecutor(m_cfg)
try:
executor.run()
except Exception as e:
logger.exception("Executor failed during execution")
try:
# 尝试更新状态,如果失败仅打印警告,不抛出异常
executor.update_db("FAILED")
except Exception as db_err:
logger.warning(f"⚠️ Failed to update DB status (real error masked by DB issue): {db_err}")
# 重新抛出原始业务异常,以便 Ray 捕获并记录到 Dashboard
raise e
executor.update_db("COMPLETED")
@@ -0,0 +1,50 @@
import os
from deepdata.common.utils import is_k8s
from deepdata.scheduler import cmd_scheduler
from deepdata.scheduler import ray_job_scheduler
from loguru import logger
# xuhuanqing
async def submit(task_id, config_path, retry_count: int = 0):
current_dir = os.path.dirname(__file__)
executor_script = os.path.join(current_dir, 'deepdata_executor.py')
# 🟢 关键修复:显式传递环境变量给 Ray 任务
runtime_env = {
"env_vars": {
"PG_HOST": os.getenv("PG_HOST", "deepdata-database"),
"PG_PORT": os.getenv("PG_PORT", "5432"),
"PG_USER": os.getenv("PG_USER", "postgres"),
"PG_PASSWORD": os.getenv("PG_PASSWORD", "password"), # 确保传递正确密码
"PG_DATABASE": os.getenv("PG_DATABASE", "deepdata"),
}
}
ray_enabled = os.getenv("ENABLE_RAY_EXECUTION", "false").lower() == "true"
if ray_enabled:
log_path = f"/flow/{task_id}/output.log" if retry_count == 0 else f"/flow/{task_id}/output.log.{retry_count}"
try:
# await ray_job_scheduler.submit(task_id, executor_script, f"--config_path={config_path}", log_path=log_path)
# 🟢 将 runtime_env 传入 submit 方法
await ray_job_scheduler.submit(
task_id,
executor_script,
f"--config_path={config_path}",
log_path=log_path,
runtime_env=runtime_env # 传入环境变量
)
return
except Exception as e:
logger.warning(f"Ray 集群不可用,降级为本地执行: {e}")
# 降级:走旧版 Runtime / 本地子进程
await cmd_scheduler.submit(
task_id,
cmd_args=["python", executor_script, f"--config_path={config_path}"]
)
def cancel(task_id):
if not is_k8s():
return cmd_scheduler.cancel_task(task_id)
return ray_job_scheduler.cancel_task(task_id)
@@ -0,0 +1,140 @@
import json
import os
import shutil
import time
from pathlib import Path
from typing import Dict
from deepdata.common.utils.file_scanner import FileScanner
import ray
from jsonargparse import dict_to_namespace
from loguru import logger
from deepdata.common.utils import check_valid_path
from deepdata.sql_manager.persistence_atction import TaskInfoPersistence
class RayExecutor:
"""
基于Ray的执行器.
1. 当前仅支持Mapper,Filter类型的算子。
2. 当前仅加载json文件类型的数据集。
"""
def __init__(self, cfg=None, meta=None):
if isinstance(cfg, Dict):
self.cfg = dict_to_namespace(cfg)
else:
logger.error(f"Please set param: cfg as type Dict, but given cfg as type {type(cfg).__name__}")
raise TypeError(f"To params cfg, Dict type is required, but type {type(cfg).__name__} is given!")
self.cfg.process = cfg.get('process', None)
self.meta = meta
# init ray
logger.info('Initing Ray ...')
ray.init()
def load_meta(self, line):
meta = json.loads(line)
if meta.get("fileId"):
meta["sourceFileId"] = meta.get("fileId")
if meta.get("fileName"):
meta["sourceFileName"] = meta.get("fileName")
if meta.get("fileType"):
meta["sourceFileType"] = meta.get("fileType")
if meta.get("fileSize"):
meta["sourceFileSize"] = meta.get("fileSize")
else:
meta["sourceFileSize"] = 0
if not meta.get("totalPageNum"):
meta["totalPageNum"] = 0
if not meta.get("extraFilePath"):
meta["extraFilePath"] = None
if not meta.get("extraFileType"):
meta["extraFileType"] = None
meta["dataset_id"] = self.cfg.dataset_id
return meta
def load_dj_meta(self, line):
meta = json.loads(line)
filepath = ""
file = ""
if meta.get("images"):
if isinstance(meta["images"], list):
filepath = meta["images"][0]
file = Path(filepath)
del meta["images"]
elif meta.get("audios"):
if isinstance(meta["audios"], list):
filepath = meta["audios"][0]
file = Path(filepath)
del meta["audios"]
elif meta.get("videos"):
if isinstance(meta["videos"], list):
filepath = meta["videos"][0]
file = Path(filepath)
del meta["videos"]
if filepath and file:
filename = f"{Path(meta['fileName']).stem}{file.suffix}"
meta["fileName"] = filename
meta["filePath"] = f"/dataset/{self.cfg.dataset_id}/{filename}"
meta["fileType"] = file.suffix[1:]
meta["fileSize"] = file.stat().st_size
os.makedirs(f"/dataset/{self.cfg.dataset_id}", exist_ok=True)
shutil.move(filepath, f"/dataset/{self.cfg.dataset_id}/{filename}")
return {k: v for k, v in meta.items() if not (isinstance(k, str) and k.startswith('_'))}
def run(self):
pass
def load_dataset(self, jsonl_file_path = None):
retry = 0
dataset = None
if jsonl_file_path is None:
jsonl_file_path = self.cfg.dataset_path
while True:
if check_valid_path(jsonl_file_path):
with open(jsonl_file_path, "r", encoding='utf-8') as meta:
lines = meta.readlines()
dataset = ray.data.from_items([self.load_meta(line) for line in lines])
break
if retry < 5:
retry += 1
time.sleep(retry)
continue
else:
logger.error(f"can not load dataset from dataset_path")
raise RuntimeError(f"Load dataset Failed!, dataset_path: {self.cfg.dataset_path}.")
return dataset
def load_dj_dataset(self, jsonl_file_path = None):
retry = 0
dataset = None
if jsonl_file_path is None:
jsonl_file_path = self.cfg.dataset_path
while True:
if check_valid_path(jsonl_file_path):
with open(jsonl_file_path, "r", encoding='utf-8') as meta:
lines = meta.readlines()
dataset = ray.data.from_items([self.load_dj_meta(line) for line in lines])
break
if retry < 5:
retry += 1
time.sleep(retry)
continue
else:
logger.error(f"can not load dataset from dataset_path")
raise RuntimeError(f"Load dataset Failed!, dataset_path: {self.cfg.dataset_path}.")
return dataset
def update_db(self, status):
task_info = TaskInfoPersistence()
task_info.update_result(self.cfg.dataset_id, self.cfg.instance_id, status)
def scan_files(self):
scanner = FileScanner(self.cfg.dataset_id)
scanner.scan_and_process(self.cfg.export_path)
+64
View File
@@ -0,0 +1,64 @@
[project]
name = "deepdata"
dynamic = ["version"]
description = "Data Processing for and with Foundation Models."
authors = [
{ name = "Huawei deepdata team" }
]
license = { text = "Apache-2.0" }
requires-python = ">=3.10, <=3.12"
urls = { repository = "https://gitea.apescale.com/figmar/deepdata" }
classifiers = [
"License :: OSI Approved :: Apache Software License",
"Programming Language :: Python :: 3",
"Operating System :: OS Independent"
]
# Core dependencies
dependencies = [
"fastapi>=0.123.9",
"jsonargparse>=4.44.0",
"loguru>=0.7.3",
"opencv-python-headless>=4.12.0.88",
"ray[data,default]>=2.52.1",
"unstructured[csv,docx,pptx,xlsx,pdf,md]==0.18.15",
"uvicorn[standard]>=0.38.0",
"sqlalchemy>=2.0.44",
"psycopg2-binary>=2.9.11",
"requests>=2.32.0",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.version]
path = "deepdata/__init__.py"
[tool.hatch.build.targets.wheel]
packages = ["deepdata"]
include = ["pyproject.toml"]
[tool.hatch.build]
include = ["pyproject.toml"]
[tool.hatch.build.targets.wheel.shared-data]
"pyproject.toml" = "pyproject.toml"
[tool.flake8]
per-file-ignores = [
"*/__init__.py: F401"
]
max-line-length = 120
extend-ignore = [
"E203", # whitespace before ':' (black handles this)
"E501", # line too long (black handles this)
"BLK100", # black would make changes (black handles this)
]
[tool.black]
line-length = 120
target-version = ['py310']
[tool.isort]
profile = "black"