Metadata-Version: 2.4
Name: roal
Version: 0.7.0
Summary: Roal 框架 - 纯数据流转与任务编排框架
License: MIT
Keywords: pipeline,data-flow,orchestration,async,task-scheduler
Author: RestRegular
Author-email: author@example.com
Requires-Python: >=3.9,<4.0
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Provides-Extra: aio
Provides-Extra: full
Provides-Extra: kafka
Provides-Extra: postgres
Provides-Extra: redis
Requires-Dist: aiofiles (>=23.0.0,<24.0.0) ; extra == "aio" or extra == "full"
Requires-Dist: aiokafka (>=0.9.0,<0.10.0) ; extra == "kafka" or extra == "full"
Requires-Dist: asyncpg (>=0.29.0,<0.30.0) ; extra == "postgres" or extra == "full"
Requires-Dist: msgpack (>=1.0.0,<2.0.0)
Requires-Dist: prometheus_client (>=0.20.0,<0.21.0)
Requires-Dist: pyyaml (>=6.0,<7.0)
Requires-Dist: redis (>=5.0.0,<6.0.0) ; extra == "redis" or extra == "full"
Requires-Dist: structlog (>=24.0.0,<25.0.0)
Requires-Dist: uvloop (>=0.19.0,<0.20.0) ; (sys_platform != "win32") and (extra == "full")
Project-URL: Bug Tracker, https://github.com/RestRegular/roal/issues
Project-URL: Changelog, https://github.com/RestRegular/roal/blob/main/worklogs
Project-URL: Documentation, https://github.com/RestRegular/roal#readme
Project-URL: Homepage, https://github.com/RestRegular/roal
Project-URL: Repository, https://github.com/RestRegular/roal
Description-Content-Type: text/markdown

# Roal 框架

## 框架概述

**Roal** = **R**outing + **O**rchestration + **A**nalytics + **L**ogistics

代表框架的核心能力：数据路由、流程编排、分析处理、物流式流转。

Roal 是一个**纯数据流转与任务编排框架**，采用 Python 语言实现。

**核心信条**：框架不关心数据从哪来、到哪去、长什么样，只负责在任务之间传递数据、编排执行顺序、管理任务生命周期。

**开发者拥有绝对自由**：
- 数据源可以是文件、数据库、消息队列、HTTP 请求、硬件传感器、随机数生成器
- 数据格式可以是字符串、JSON、Protobuf、Avro、二进制字节流、Pandas DataFrame、NumPy 数组
- 处理逻辑可以是任何 Python 代码

**框架职责边界**：开发者写好 Task，框架负责串联执行。

## 核心概念

### Task（任务）

Task 是框架中最基础的抽象，代表一个独立的处理单元。

**开发者视角**：Task 就是一个可调用对象（函数、lambda、类的 `__call__` 方法），接收一个输入，产出一个输出。输入输出的类型完全由开发者决定。

**框架视角**：Task 是一个黑盒，框架只知道它有输入和输出，不关心内部实现。框架负责在合适的时机调用它，并把输出传递给下一个 Task。

**定义方式**：
- 普通函数
- 异步函数（async def）
- 生成器函数（yield）
- 实现了 `__call__` 的类实例
- 使用 `@task` 装饰器增强的任意可调用对象

### Pipeline（流水线）

Pipeline 是多个 Task 的有序组合，定义了数据从产生到消费的完整路径。

**构成元素**：
- **节点**：每个 Task 是一个节点
- **边**：表示数据流向，从上游 Task 的输出指向下游 Task 的输入
- **路由规则**：定义数据如何从上游分发到下游

**支持的拓扑结构**：
- **串联**：A → B → C
- **分流**：A 的输出同时发给 B 和 C（广播）
- **合并**：B 和 C 的输出都发给 D
- **条件分支**：根据数据内容决定发送给哪个下游
- **扇出扇入**：一对多再聚合
- **哈希路由**：根据数据字段的哈希值选择下游

### TaskContext（任务上下文）

TaskContext 是框架暴露给开发者的服务接口，让 Task 在执行时能够获取框架提供的能力。

**提供的服务**：
- **配置获取**：获取 Task 的配置参数，支持类型转换和默认值
- **状态存储**：可选的 KV 存储，用于保存跨数据记录的 Task 状态
- **指标上报**：向框架上报自定义指标（计数器、直方图、仪表）
- **日志记录**：自动携带流水线 ID、Task 名称、数据 ID 的结构化日志
- **元数据访问**：获取当前 Pipeline 的名称、运行 ID、启动时间等信息
- **数据属性**：获取当前数据的元数据（来源、时间戳、序列号）

## 安装

### 基本安装

```bash
pip install roal
```

### 安装可选依赖

```bash
# 安装 Redis 支持
pip install roal[redis]

# 安装异步文件 I/O 支持
pip install roal[aio]

# 安装 Kafka 支持
pip install roal[kafka]

# 安装 PostgreSQL 支持
pip install roal[postgres]

# 安装所有可选依赖
pip install roal[redis,aio,kafka,postgres]
```

## 快速开始

### 定义 Task

```python
from roal import task, TaskContext
import json

@task(name="parse_json", workers=4)
async def parse_json(data: bytes, ctx: TaskContext) -> dict:
    return json.loads(data)

@task
async def filter_errors(record: dict, ctx: TaskContext) -> dict:
    if record.get("level") == "error":
        return record
    return None  # 过滤非错误记录

@task
async def send_alert(error: dict, ctx: TaskContext) -> None:
    print(f"Alert: {error['message']}")
```

### 组装 Pipeline

```python
from roal import Pipeline

pipeline = Pipeline("log_processor")

pipeline.source(read_logs) \  # 假设 read_logs 是一个数据源 Task
    .then(parse_json) \    
    .fork() \
        .to(filter_errors, send_alert) \
        .to(extract_fields, store_to_db) \
    .merge() \
    .then(generate_stats) \
    .sink(write_output)

# 运行 Pipeline
pipeline.run()
```

## 核心特性

1. **数据血缘追踪**：框架为每一条数据记录分配全局唯一的 ID，并记录它经过的每一个 Task。
2. **可观测性**：自动采集吞吐量、延迟、队列深度等指标，支持 Prometheus 集成。
3. **断点续传**：框架定期保存每个 Task 的消费位点，支持从故障中恢复。
4. **动态配置**：支持运行时修改 Task 配置，无需重启 Pipeline。
5. **测试支持**：提供内存数据源和数据汇，简化单元测试。

## 典型应用场景

- **实时日志分析**：实时监控应用日志，错误日志发送告警，访问日志统计 QPS。
- **数据 ETL 管道**：从业务库抽取数据，清洗转换后写入数仓，同时更新缓存。
- **物联网数据处理**：接收设备传感器数据，异常检测，持久化，下发指令。
- **AI 推理流水线**：接收图片，预处理，模型推理，后处理，返回结果。
- **事件驱动工作流**：用户操作触发一系列后续处理。

## 技术选型

- **核心依赖**：asyncio, uvloop, structlog, prometheus_client, PyYAML
- **Python 版本**：3.9+
- **操作系统**：Linux / macOS / Windows

## 部署形态

- **嵌入式部署**：作为 Python 库直接嵌入现有应用。
- **独立服务**：提供可选的 HTTP API 服务层。
- **容器化部署**：提供官方 Docker 镜像。
- **Kubernetes 部署**：支持 K8s 部署，利用 Horizontal Pod Autoscaler 实现自动扩缩容。

## 与现有框架对比

| 特性 | Roal | Apache Flink | Apache Airflow | Prefect | Bytewax |
|:---|:---|:---|:---|:---|:---|
| **语言** | Python | Java/Python | Python | Python | Python |
| **编程模型** | 纯函数式 Task | 富算子 | DAG 定义 | 函数式 Flow | 数据流 |
| **部署复杂度** | 极低，pip install | 高，需要集群 | 中，需要调度器 | 低 | 低 |
| **学习曲线** | 几分钟 | 数周 | 数天 | 数小时 | 数小时 |
| **实时处理** | 原生支持 | 原生支持 | 分钟级调度 | 支持 | 原生支持 |
| **状态管理** | 可选，开发者控制 | 强制，框架管理 | 有限（XCom） | 支持 | 支持 |
| **Python 生态集成** | 无缝 | 通过 PyFlink | 一般 | 良好 | 良好 |
| **适用场景** | 通用数据编排 | 大规模流计算 | 定时任务调度 | 数据工程 | 流处理 |

## 设计原则

1. **简单至上**：API 应该让开发者在 5 分钟内理解并开始使用
2. **零侵入**：不强制继承基类，普通函数即可成为 Task
3. **显式优于隐式**：框架行为可预测，避免魔法
4. **渐进式复杂度**：简单场景只需几行代码，复杂场景提供完整扩展能力
5. **Pythonic**：遵循 Python 社区惯例，利用语言特性简化设计
6. **测试友好**：提供一流测试支持，鼓励 TDD

## 项目结构

```
roal/
├── pyproject.toml                 # 项目配置和依赖
├── README.md
├── docs/                          # 文档
│   ├── getting-started.md
│   ├── core-concepts.md
│   ├── advanced/
│   └── api-reference/
├── roal/                          # 核心源码
│   ├── __init__.py
│   ├── api/                       # 公开 API
│   │   ├── __init__.py
│   │   ├── task.py               # Task 协议和装饰器
│   │   ├── pipeline.py           # Pipeline 类
│   │   └── context.py            # TaskContext 类
│   ├── core/                      # 内部实现
│   │   ├── dag.py                # DAG 构建和拓扑排序
│   │   ├── router.py             # 路由策略实现
│   │   └── executor.py           # 执行引擎
│   ├── runtime/                   # 运行时
│   │   ├── worker.py             # Worker 协程管理
│   │   ├── queue.py              # 异步队列封装
│   │   └── backpressure.py       # 背压控制
│   ├── state/                     # 状态管理
│   │   ├── base.py               # StateStore 抽象
│   │   ├── memory.py             # 内存存储
│   │   ├── redis.py              # Redis 存储
│   │   └── file.py               # 文件存储
│   ├── checkpoint/                # 断点续传
│   │   ├── manager.py
│   │   └── stores.py
│   ├── monitoring/                # 监控指标
│   │   ├── metrics.py
│   │   ├── tracing.py
│   │   └── prometheus.py
│   └── testing/                   # 测试工具
│       ├── harness.py
│       └── fixtures.py
├── examples/                      # 示例代码
│   ├── wordcount/
│   ├── log_analysis/
│   ├── etl_pipeline/
│   └── iot_processing/
└── tests/                         # 测试
    ├── unit/
    ├── integration/
    └── performance/
```

## 版本路线图

| 版本 | 里程碑 | 核心功能 |
|:---|:---|:---|
| **0.1.0** | MVP | 基础 Task 和 Pipeline，支持串联和分支，asyncio 执行器 |
| **0.2.0** | 路由增强 | 条件路由、哈希路由、广播路由，Worker 并发 |
| **0.3.0** | 状态管理 | StateStore 抽象，内存/Redis 实现 |
| **0.4.0** | 可观测性 | 指标收集，Prometheus 集成，结构化日志 |
| **0.5.0** | 容错能力 | 断点续传，重试策略，死信队列 |
| **0.6.0** | 动态能力 | 运行时配置更新，动态扩缩容 |
| **0.7.0** | 进程扩展 | 多进程支持，CPU 密集型任务隔离 |
| **0.8.0** | API 服务 | HTTP API，Web UI 基础 |
| **0.9.0** | 生态集成 | 官方连接器（Kafka、Redis、PostgreSQL） |
| **1.0.0** | 稳定版 | API 稳定，生产就绪，完整文档 |

## 贡献指南

Roal 采用开放治理模式，欢迎社区贡献：

- **代码贡献**：Fork → Feature Branch → Pull Request
- **文档改进**：直接提交 PR 到 docs 目录
- **Bug 报告**：使用 GitHub Issues，附带最小复现示例
- **功能建议**：先在 Discussions 中讨论，达成共识后再实现

**开发环境搭建**：

```bash
git clone https://github.com/RestRegular/roal.git
cd roal
python -m venv venv
source venv/bin/activate
pip install -e ".[dev]"
pytest
```

---

*Roal —— 让数据流转像呼吸一样自然*

