• 数据工程
• 大数据
• 批处理

导读：本文以 smallpond 文档 docs/source/api/tasks.rst 为骨架，系统讲解 smallpond 任务执行系统的三大核心组件——RuntimeContext（任务运行上下文）、ExecutionPlan（任务 DAG 执行计划）以及从Task基类派生的完整任务类体系。你将掌握如何手工创建一个运行时上下文、如何通过Planner将逻辑计划转换为可执行的任务图，以及各任务类在数据摄取、SQL 计算、哈希分区、流式处理、数据落盘等环节中的职责与底层实现。本文面向希望深入理解 smallpond 调度内核，或需要自定义任务类型的开发者。

一、任务系统总览：从逻辑计划到执行计划

smallpond 是一个构建在 DuckDB 与 3FS 之上的轻量级数据处理框架。其核心执行模型是"逻辑计划（LogicalPlan）→ 执行计划（ExecutionPlan）→ 调度执行"三段式流水线：

1. 逻辑计划：由用户在会话（Session）中通过Context和各Node构建，描述"做什么"（数据源、分区、SQL、投影、落盘等节点构成的有向无环图）。
2. 执行计划：由Planner（smallpond/logical/planner.py）对逻辑计划做copy.deepcopy后逐节点访问，把每个逻辑节点物化为一个或多个Task，最终组装成一棵以RootTask为根的任务 DAG，即ExecutionPlan。
3. 调度执行：JobManager（smallpond/execution/manager.py）负责创建RuntimeContext、提交执行计划，随后由Scheduler与Executor负责任务的调度与执行；任务失败时按max_retry_count重试。

文档 docs/source/api/tasks.rst 给出了手工构建这一流水线的最小示例，也是本文贯穿始终的主线代码：

# create a runtime context runtime_ctx = RuntimeContext(JobId.new(), data_root) runtime_ctx.initialize(socket.gethostname(), cleanup_root=True) # create a logical plan plan = create_logical_plan() # create an execution plan planner = Planner(runtime_ctx) exec_plan = planner.create_exec_plan(plan)

之后便可将exec_plan交给调度器执行（见 docs/source/api/execution.rst 的 "Submit a Job" 一节）。

二、RuntimeContext：一次任务运行的配置与状态容器

RuntimeContext（smallpond/execution/task.py）被文档定义为 "The configuration and state for a running job"，它是任务系统中最核心的上下文对象，被创建后通过 pickle 序列化分发到各个执行节点，因此它必须是完全可序列化的。

2.1 构造函数与核心参数

RuntimeContext( job_id: JobId, job_time: datetime, data_root: str, *, num_executors: int = 1, random_seed: int = None, env_overrides: Dict[str, str] = None, bind_numa_node=False, enforce_memory_limit=False, max_usable_cpu_count: int = 1024, max_usable_gpu_count: int = 1024, max_usable_memory_size: int = 16 * TB, secs_wq_poll_interval: float = 0.5, secs_executor_probe_interval: float = 30, max_num_missed_probes: int = 6, fault_inject_prob=0.0, enable_profiling=False, enable_diagnostic_metrics=False, remove_empty_parquet=False, skip_task_with_empty_input=False, shared_log_root: Optional[str] = None, console_log_level="INFO", file_log_level="DEBUG", disable_log_rotation=False, output_path: Optional[str] = None, )

关键参数的作用与底层影响：

2.2 initialize 与目录布局

initialize(exec_id, root_exist_ok=True, cleanup_root=False)是任务运行前的必经步骤，依次完成：

• _make_dirs：创建config_root、queue_root、output_root、staging_root、temp_root、log_root以及 Ray 执行器专用的started_tasks/completed_tasks目录；若cleanup_root=True则先删除已存在的作业根目录。
• _init_logs：注册 loguru 的 stdout、文件、用户日志（{exec_id}-user.log）三类 handler，日志轮转默认 "100 MB / 保留 5 份"（可通过disable_log_rotation关闭）。
• _init_envs：将sys.setrecursionlimit(100000)，并把sys.prefix/lib注入LD_LIBRARY_PATH，最后应用env_overrides。

RuntimeContext还提供了若干路径属性，执行计划与调度状态都以 pickle 形式落盘在这些位置：runtime_ctx_path、logcial_plan_path、logcial_plan_graph_path、exec_plan_path、sched_state_path。JobManager.run()会依次 dump 运行时上下文、逻辑计划（并渲染 PNG 图）、执行计划，再启动调度器（manager.py）。

2.3 配套标识符与统计类型

文档 autosummary 中与RuntimeContext同级的四个辅助类型也定义在 task.py：

• JobId（继承uuid.UUID）：作业唯一标识，JobId.new()以uuid4生成。
• TaskId（继承int）：任务的全局递增编号，由RuntimeContext.new_task_id()分配，其__str__输出为 6 位补零格式（如000001）。
• TaskRuntimeId（frozen dataclass）：运行时任务标识，由(id, epoch, retry)三元组构成，str形如000001.0.0，其中epoch为调度轮次、retry为已重试次数——这正是任务输出的目录名后缀（output_filename中的{job_id}.{runtime_id}），保证每次重试写入不同目录，天然支持幂等重试与 Ray 执行器的断点恢复。
• PartitionInfo：分区信息，字段为(index, npartitions, dimension)。顶层维度为@toplevel@，默认数据分区维度为DATA_PARTITION_COLUMN_NAME；__str__输出如dimension[0/10]。它同时定义了排序与相等比较，使任务可以按(dimension, index)有序排列，便于消费者按分区索引对齐。
• PerfStats：命名元组，包含(cnt, sum, min, max, avg, p50, p75, p95, p99)九个统计量，JobManager收集各任务perf_metrics后按此聚合为逻辑节点的性能统计。

三、ExecutionPlan：任务构成的有向无环图

ExecutionPlan（task.py）在源码中被描述为 "A directed acyclic graph (DAG) of tasks"。它的构造需要三要素：RuntimeContext、RootTask和原始逻辑计划LogicalPlan（用于执行完成后回填各节点的性能统计）。

exec_plan = ExecutionPlan(ctx, root_task, logical_plan)

3.1 关键属性与能力

• root_task：任务 DAG 的根节点，其input_deps挂载全部最终汇任务；str(exec_plan)会以缩进树的形式递归打印整棵任务图。
• tasks/leaves：通过iter_tasks深度优先遍历（以task.key去重）得到全部任务字典；leaves即无输入依赖的任务（典型如DataSourceTask）。
• successful：root_task.status == WorkStatus.SUCCEED，是整个作业成功与否的总开关，测试中常以此断言（见 tests/test_execution.py）。
• final_output/final_output_path：定位唯一的DataSinkTask（is_final_node=True），暴露最终结果数据集及其路径。
• named_outputs/get_output(name)：按output_name聚合各命名任务的输出数据集，同名数据集会被merge合并。
• analyzed_logical_plan：执行成功后，把各任务的perf_metrics通过node.get_perf_stats回填到逻辑计划节点，形成带统计信息的最终逻辑计划。

3.2 Planner 如何构建执行计划

Planner（smallpond/logical/planner.py）继承LogicalPlanVisitor，用visit_*方法把每种逻辑节点翻译为任务组（TaskGroup），并通过node_to_tasks做记忆化以避免重复展开：

• 数据源节点 → 单个DataSourceTask（叶子任务）。
• 分区节点 → 按nested与否生成 Producer/Consumer 任务对：PartitionProducerTask系列负责把输入切分成 npartitions 份，PartitionConsumerTask按分区索引对齐并合并。
• 计算节点（Python / Arrow / SQL）→ 通过broadcast_input_deps把输入依赖按分区广播对齐，为每个分区生成一个任务实例。
• 每个带output_name的逻辑节点都会挂接一个DataSinkNode；最后以RootNode收拢所有 sink 节点并生成RootTask。

create_exec_plan的最后还做了两处工程化优化（planner.py）：若根节点是ConsolidateNode且其输入是单分区节点，则以 manifest 文件替代复制文件；否则为根节点生成FinalResults类型的最终DataSinkTask。

四、Task 基类：任务的生命周期与核心约定

Task（task.py）是所有任务类型的基类，也是 smallpond 最小的执行单元。文档注释明确了它的设计约定：

Task is the basic unit of work in smallpond. Each task represents a specific operation that takes a series of input datasets and produces an output dataset. Tasks can depend on other tasks, forming a directed acyclic graph (DAG). Tasks can specify resource requirements such as CPU, GPU, and memory limits. Tasks should be idempotent. They can be retried if they fail.

4.1 任务对象的一生（Lifetime）

一个任务对象完整经历以下阶段，且全程依赖 pickle 在调度节点与工作节点间传输：

1. 规划期实例化：在调度节点上由Planner创建。
2. 序列化分发：被 pickle 后发送到工作节点（Executor）。
3. initialize()：准备执行——注入故障检查、按 CPU 配额推算内存限制（usable_memory_size * cpu_limit / usable_cpu_count）、创建运行时输出目录、设置随机种子、按cpu_limit限定 Arrow / OMP / Polars 线程数。
4. run()：真正执行任务逻辑，各子类在此实现具体行为。
5. finalize()/cleanup()：前者把运行时输出目录原子重命名到最终输出目录（os.rename），汇总perf_metrics与资源用量（ru_maxrss、ru_utime、ru_stime）；后者清理剖析器与复杂属性，保证任务对象可再次序列化回调度节点。
6. 返回调度节点：任务状态与指标随调度状态（sched_state）持久化，作业可据此断点续跑。

4.2 资源约束与分区

Task.__init__接受cpu_limit（默认 1）、gpu_limit（默认 0）、memory_limit（默认按 CPU 配额从usable_memory_size推算），并校验partition_infos中必须包含@toplevel@维度、分区维度不得重复。任务通过parquet_kv_metadata_str/bytes把自己的task_key、task_id、node_id、job_id以及分区信息写入输出 parquet 文件的 KV metadata，下游消费者据此重建数据集与分区归属。

4.3 容错与随机数约定

任务必须是幂等的（可安全重试）：每次重试的TaskRuntimeId不同，输出目录天然隔离；initialize()与finalize()均调用inject_fault()以支持故障注入测试。任务随机数由random_seed_bytes = task_id + ctx.random_seed派生，numpy_random_gen与python_random_gen保证同一任务每次重试产生相同的随机序列。

4.4 调试利器：Task Local Runner

task.py自带main()入口，可从 pickle 文件加载单个或多个任务在本地直接执行（python -m smallpond.execution.task <pickle_path> [-t task_id] [-r retry_count] [-o output_path] [-l log_level]），是排查任务逻辑的实用工具。

五、任务类体系：从数据摄取到最终落盘

文档 autosummary 列出的任务类可分为五条职责主线，全部定义在 smallpond/execution/task.py：

5.1 数据源与图结构任务（在调度节点执行）

这些任务exec_on_scheduler=True，只做元数据操作、不搬移数据：

• DataSourceTask：叶子任务，直接持有DataSet；若数据集是SqlQueryDataSet/ArrowTableDataSet，会先物化为 parquet 文件再包装为ParquetDataSet。
• RootTask：执行计划根节点，run()恒为 True，仅用于聚合状态。
• MergeDataSetsTask：把多个同构数据集merge为一个。
• SplitDataSetTask：按文件切分数据集，供分区 Producer 使用。
• ProjectionTask：在数据集上应用列裁剪（columns）、生成列（generated_columns，仅 parquet）与union_by_name语义。
• DataSinkTask：作业的最终出口，支持四种type：link_manifest（同挂载点硬链接 + 写.MANIFEST.txt）、copy（跨挂载点复制）、link_or_copy（同挂载点链接、否则复制）、manifest（仅输出文件清单）。复制使用 16MB 缓冲的shutil.copyfileobj（源码注释明确避免 Linux sendfile 的 64KB 固定 I/O 低吞吐问题），manifest 先写临时文件再原子重命名。默认清单文件名为.MANIFEST.txt。

5.2 分区 Producer/Consumer 家族

• PartitionProducerTask（抽象基类）：负责把输入数据集切成npartitions份，产出partitioned_datasets列表。其finalize()有两点关键保证：分区数必须等于npartitions；对空分区会按首个非空分区的 schema 创建带分区元数据的"空文件"（parquet 用ParquetWriter写入带 KV 元数据的空 schema），确保下游消费者至少有一个可读文件。
• PartitionConsumerTask：消费者，按last_partition.index从各输入的PartitionedDataSet中取出对应分区并merge合并；可选移除空 parquet 文件。
• RepeatPartitionProducerTask：把同一数据集重复 npartitions 次（广播语义）。
• EvenlyDistributedPartitionProducerTask：按文件（partition_by_files）或按行（partition_by_rows，仅 parquet）均匀切分，支持random_shuffle。
• UserDefinedPartitionProducerTask：接受用户提供的partition_func(ctx, dataset) -> List[DataSet]自定义分区逻辑（执行后立即置空函数以便序列化）。
• LoadPartitionedDataSetProducerTask：从带data_partition_column列的 parquet 数据加载既有分区，支持 Hive 风格目录分区。
• RangePartitionTask：范围分区任务占位基类。

5.3 SQL / Arrow / Pandas 计算任务

• ExecSqlQueryMixin（混入基类）：提供 DuckDB 连接准备（prepare_connection，设置threads、memory_limit、temp_directory、arrow_large_buffer_size、preserve_insertion_order=false等）、输入数据集注册为视图（create_input_views，视图名形如input_{id}_{seq:06d}）、查询执行与 DuckDB profiling 指标抽取（exec_query）。duckdb.OutOfMemoryException会被转换为框架的OutOfMemory异常。
• SqlEngineTask：执行多条 SQL（以{0}、{1}占位引用输入视图），支持per_thread_output、materialize_output、materialize_in_memory、batched_processing等模式；最终通过COPY ... TO写出 parquet（可带 KV 元数据、压缩、row group 配置、PER_THREAD_OUTPUT、FILENAME_PATTERN）。
• ArrowComputeTask：一次性把输入读为List[arrow.Table]，用户process_func(ctx, tables) -> Table处理后再批量写 parquet；支持use_duckdb_reader、字典编码、ZSTD 压缩与行组大小自适应（adjust_row_group_size，约束MAX_NUM_ROW_GROUPS/MAX_ROW_GROUP_BYTES）。
• ArrowStreamTask：流式任务，输入为List[arrow.RecordBatchReader]，用户处理函数逐批产出arrow.Table；输出边写边落盘，支持后台 I/O 线程（background_io_thread，配合ConcurrentIter）、内存上限推导的批大小裁剪，以及基于RuntimeState的断点 checkpoint（按secs_checkpoint_interval周期或StreamOutput.force_checkpoint触发，checkpoint 任务会 push 回执行队列）。这是处理超大数据集的主力任务。
• ArrowBatchTask：ArrowStreamTask的批处理变体，按批zip_longest对齐输入 reader，逐批调用process(ctx, tables)。
• PandasComputeTask/PandasBatchTask：分别继承 Arrow 计算/批处理任务，将输入转为pd.DataFrame交给用户函数，输出再转回arrow.Table。
• PythonScriptTask：接受process_func(ctx, List[DataSet], output_path) -> bool，面向以文件为粒度的脚本型处理，输出包装为FileSet。

5.4 哈希分区任务

• HashPartitionTask（抽象基类）：哈希分区核心，支持random_shuffle（以random()为哈希键）、shuffle_only、drop_partition_column、Hive 目录分区等选项；内部用ThreadPoolExecutor(num_workers)并行写多个分区文件，并设置自适应写缓冲（write_buffer_size，最小 128KB，过小则禁用）。finalize()校验"输入行数 == 输出行数"。
• HashPartitionDuckDbTask：DuckDB 引擎实现。分区键为CAST(hash(concat_ws('##', cols)) AS UINT64) % npartitions；先ORDER BY partition_key灌入临时表，再并行COPY出各分区文件（或按PARTITION_BY写 Hive 目录）。
• HashPartitionArrowTask：Arrow/Polars 引擎实现，用 polarshash_rows(*fixed_rand_seeds)计算哈希并partition_by分组；源码注释特别警告：随机种子改变会生成完全不同的哈希分区。
• 工厂方法HashPartitionTask.create(engine_type, ...)按"duckdb"/"arrow"选择实现。

5.5 任务类与逻辑节点的对应关系

下表是文档 tasks.rst 所列任务类在逻辑层（smallpond/logical/node.py）中的"生产方"：

六、端到端实践：从 RuntimeContext 到调度执行

6.1 编程式提交（JobManager 方式）

文档 docs/source/api/execution.rst 推荐的正式用法是通过Driver作为脚本入口，其内部是JobManager的薄封装：

from smallpond.execution.driver import Driver if __name__ == "__main__": driver = Driver() # add your own arguments driver.add_argument("-i", "--input_paths", nargs="+") driver.add_argument("-n", "--npartitions", type=int, default=10) # build and run logical plan plan = my_pipeline(**driver.get_arguments()) driver.run(plan)

JobManager.run()的完整链路（manager.py）依次为：构造RuntimeContext→initialize()→ dump 运行时上下文/逻辑计划/执行计划 →Planner.create_exec_plan→ 启动执行器（platform.start_job，每个执行器以driver.py --job_id ... --data_root ... executor启动）→Scheduler.run()阻塞至作业完成并返回exec_plan。若上次调度状态文件（sched_state_path）存在，还会加载并以sched_epoch + 1续跑，实现断点恢复。

6.2 测试中的最小执行回路

测试框架（tests/test_fabric.py）展示了RuntimeContext + Planner + ExecutionPlan + Scheduler/Executor的最小回路：execute_plan启动一个调度器进程与num_executors个执行器进程，join_running_procs()等待完成后断言latest_state.success，最后返回ExecutionPlan供断言final_output。测试中构造RuntimeContext的典型方式为：

runtime_ctx = RuntimeContext( JobId.new(), datetime.now(), self.output_root_abspath, console_log_level="WARNING", ) runtime_ctx.initialize(socket.gethostname(), cleanup_root=True)

（见 tests/test_execution.py）。此外 tests/test_logical.py 直接以Planner(runtime_ctx).create_exec_plan(plan)验证逻辑→执行计划的转换正确性。

七、小结

smallpond 的任务系统遵循"上下文（RuntimeContext）—计划（ExecutionPlan）—任务（Task）"三层模型：RuntimeContext承载作业级配置、目录布局与标识符分配；Planner将逻辑计划物化为以RootTask为根的任务 DAG；各Task子类以幂等、可重试、资源受限的方式完成数据摄取、分区、计算与落盘。理解这套体系后，你既可以在Driver/JobManager之上开发完整的数据处理流水线，也可以通过继承Task或实现PartitionProducerTask扩展自定义任务类型。想进一步深入调度细节，可继续阅读 docs/source/api/execution.rst 以及Scheduler、Executor、Platform的相关文档与源码。

• 数据工程
• 大数据
• 批处理