数据工程大数据批处理【免费下载链接】smallpondA lightweight data processing framework built on DuckDB and 3FS.项目地址https://gitcode.com/gh_mirrors/sma/smallpond点击查看免费下载导读本文以 smallpond 文档 docs/source/api/tasks.rst 为骨架系统讲解 smallpond 任务执行系统的三大核心组件——RuntimeContext任务运行上下文、ExecutionPlan任务 DAG 执行计划以及从Task基类派生的完整任务类体系。你将掌握如何手工创建一个运行时上下文、如何通过Planner将逻辑计划转换为可执行的任务图以及各任务类在数据摄取、SQL 计算、哈希分区、流式处理、数据落盘等环节中的职责与底层实现。本文面向希望深入理解 smallpond 调度内核或需要自定义任务类型的开发者。一、任务系统总览从逻辑计划到执行计划smallpond 是一个构建在 DuckDB 与 3FS 之上的轻量级数据处理框架。其核心执行模型是逻辑计划LogicalPlan→ 执行计划ExecutionPlan→ 调度执行三段式流水线逻辑计划由用户在会话Session中通过Context和各Node构建描述做什么数据源、分区、SQL、投影、落盘等节点构成的有向无环图。执行计划由Plannersmallpond/logical/planner.py对逻辑计划做copy.deepcopy后逐节点访问把每个逻辑节点物化为一个或多个Task最终组装成一棵以RootTask为根的任务 DAG即ExecutionPlan。调度执行JobManagersmallpond/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_rootTrue) # 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一次任务运行的配置与状态容器RuntimeContextsmallpond/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_nodeFalse, enforce_memory_limitFalse, 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_prob0.0, enable_profilingFalse, enable_diagnostic_metricsFalse, remove_empty_parquetFalse, skip_task_with_empty_inputFalse, shared_log_root: Optional[str] None, console_log_levelINFO, file_log_levelDEBUG, disable_log_rotationFalse, output_path: Optional[str] None, )关键参数的作用与底层影响参数默认值作用job_id/job_time必填唯一标识一次作业job_time连同job_id构成作业根目录名{job_time:%Y-%m-%d-%H-%M-%S}.{job_id}data_root必填作业数据的根目录其下自动创建config、queue、output、staging、temp、log六个子目录num_executors1执行器数量用于计算分区任务的并行度num_parallel_tasks 2 * num_executors * ceil(usable_cpu_count / cpu_limit)见 planner.pyrandom_seed随机生成若为None则用os.urandom生成与任务 ID 一起决定每个任务可复现的随机数种子random_seed_bytesbind_numa_nodeFalse是否绑定 NUMA 节点开启时 CPU、内存、GPU 数量都按numa_node_count均分enforce_memory_limitFalse开启后任务会用resource.setrlimit(RLIMIT_DATA)设置软/硬内存上限1.2x / 1.5x 任务内存限制max_usable_cpu_count/gpu/memory1024 / 1024 / 16TB任务可用的资源上限最终取min(上限, 本机实际值)见usable_cpu_count、usable_memory_sizesecs_executor_probe_interval/max_num_missed_probes30 / 6执行器心跳探测间隔与最大失联次数二者相乘即执行器超时阈值secs_executor_probe_timeout也是流式任务默认的 checkpoint 间隔fault_inject_prob0.0故障注入概率用于测试任务重试与容错inject_fault()会以uniform_failure_prob与fault_inject_prob的较小者抛InjectedFaultenable_profilingFalse开启 cProfile 性能剖析耗时超过 3 秒的任务会在 finalize 时输出剖析结果enable_diagnostic_metricsFalse开启后记录输入/输出文件数量与总大小等诊断指标remove_empty_parquetFalse读取输入时移除空的 parquet 文件skip_task_with_empty_inputFalse任一输入为空时直接跳过任务skip_when_any_input_emptyshared_log_rootNone共享日志根目录用于跨节点汇聚日志console_log_level/file_log_levelINFO / DEBUG控制台与日志文件的 loguru 级别output_pathNone若指定Planner会生成类型为copy的最终DataSink见 planner.py2.2 initialize 与目录布局initialize(exec_id, root_exist_okTrue, cleanup_rootFalse)是任务运行前的必经步骤依次完成_make_dirs创建config_root、queue_root、output_root、staging_root、temp_root、log_root以及 Ray 执行器专用的started_tasks/completed_tasks目录若cleanup_rootTrue则先删除已存在的作业根目录。_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.pyJobId继承uuid.UUID作业唯一标识JobId.new()以uuid4生成。TaskId继承int任务的全局递增编号由RuntimeContext.new_task_id()分配其__str__输出为 6 位补零格式如000001。TaskRuntimeIdfrozen 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任务构成的有向无环图ExecutionPlantask.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。successfulroot_task.status WorkStatus.SUCCEED是整个作业成功与否的总开关测试中常以此断言见 tests/test_execution.py。final_output/final_output_path定位唯一的DataSinkTaskis_final_nodeTrue暴露最终结果数据集及其路径。named_outputs/get_output(name)按output_name聚合各命名任务的输出数据集同名数据集会被merge合并。analyzed_logical_plan执行成功后把各任务的perf_metrics通过node.get_perf_stats回填到逻辑计划节点形成带统计信息的最终逻辑计划。3.2 Planner 如何构建执行计划Plannersmallpond/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 基类任务的生命周期与核心约定Tasktask.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 在调度节点与工作节点间传输规划期实例化在调度节点上由Planner创建。序列化分发被 pickle 后发送到工作节点Executor。initialize()准备执行——注入故障检查、按 CPU 配额推算内存限制usable_memory_size * cpu_limit / usable_cpu_count、创建运行时输出目录、设置随机种子、按cpu_limit限定 Arrow / OMP / Polars 线程数。run()真正执行任务逻辑各子类在此实现具体行为。finalize()/cleanup()前者把运行时输出目录原子重命名到最终输出目录os.rename汇总perf_metrics与资源用量ru_maxrss、ru_utime、ru_stime后者清理剖析器与复杂属性保证任务对象可再次序列化回调度节点。返回调度节点任务状态与指标随调度状态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 Runnertask.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.py5.1 数据源与图结构任务在调度节点执行这些任务exec_on_schedulerTrue只做元数据操作、不搬移数据DataSourceTask叶子任务直接持有DataSet若数据集是SqlQueryDataSet/ArrowTableDataSet会先物化为 parquet 文件再包装为ParquetDataSet。RootTask执行计划根节点run()恒为 True仅用于聚合状态。MergeDataSetsTask把多个同构数据集merge为一个。SplitDataSetTask按文件切分数据集供分区 Producer 使用。ProjectionTask在数据集上应用列裁剪columns、生成列generated_columns仅 parquet与union_by_name语义。DataSinkTask作业的最终出口支持四种typelink_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_orderfalse等、输入数据集注册为视图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 回执行队列。这是处理超大数据集的主力任务。ArrowBatchTaskArrowStreamTask的批处理变体按批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()校验输入行数 输出行数。HashPartitionDuckDbTaskDuckDB 引擎实现。分区键为CAST(hash(concat_ws(##, cols)) AS UINT64) % npartitions先ORDER BY partition_key灌入临时表再并行COPY出各分区文件或按PARTITION_BY写 Hive 目录。HashPartitionArrowTaskArrow/Polars 引擎实现用 polarshash_rows(*fixed_rand_seeds)计算哈希并partition_by分组源码注释特别警告随机种子改变会生成完全不同的哈希分区。工厂方法HashPartitionTask.create(engine_type, ...)按duckdb/arrow选择实现。5.5 任务类与逻辑节点的对应关系下表是文档 tasks.rst 所列任务类在逻辑层smallpond/logical/node.py中的生产方任务类对应的逻辑节点 / 产生场景DataSourceTaskDataSourceNode数据源RootTask/ExecutionPlanRootNode→Planner.visit_root_nodeDataSinkTaskDataSinkNode含FinalResults汇任务MergeDataSetsTask/SplitDataSetTask分区节点的合并/切分辅助任务create_merge_task/create_split_taskPartitionProducerTask系列PartitionNode.create_producer_taskPartitionConsumerTaskPartitionNode.create_consumer_taskArrowComputeTask/ArrowStreamTask/ArrowBatchTaskArrowComputeNode/ArrowStreamNodePandasComputeTask/PandasBatchTask上述 Arrow 节点的 Pandas 变体SqlEngineTaskSqlEngineNodePythonScriptTaskPythonScriptNodeHashPartitionTask系列PartitionNode的哈希分区变体ProjectionTaskProjectionNodeRangePartitionTaskPartitionNode的范围分区变体六、端到端实践从 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, typeint, default10) # 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_levelWARNING, ) runtime_ctx.initialize(socket.gethostname(), cleanup_rootTrue)见 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的相关文档与源码。赞分享数据工程大数据批处理【免费下载链接】smallpondA lightweight data processing framework built on DuckDB and 3FS.项目地址https://gitcode.com/gh_mirrors/sma/smallpond点击查看免费下载相关推荐CAMEL 任务系统深度解析用 Task 与 TaskManager 构建可分解、可编排的多智能体任务CAMEL 任务系统深度解析用 Task 与 TaskManager 构建可分解、可编排的多智能体任务 本篇技术指南聚焦 CAMEL 多智能体框架当前仓库人工智能大模型AI AgentAgent 框架多智能体工具调用MCP ClientsRAG模型评测数据生成Task任务调度工具现代化构建系统的架构实践深度解析Task任务调度工具现代化构建系统的架构实践深度解析 Task是一个基于Go语言开发的轻量级任务调度和构建工具旨在提供比传统Makefile更简单易用的替代构建工具开发工具CLItsParticles 粒子阴影Particles Shadow配置详解为每个粒子叠加 Canvas 投影效果tsParticles 粒子阴影Particles Shadow配置详解为每个粒子叠加 Canvas 投影效果 导读 本文围绕 tsParticles 引构建工具开发工具CLI上一篇音乐搜索器部署指南从零开始搭建完整的音乐搜索服务下一篇Reactive Resume 完整指南免费开源 AI 简历生成器55 种语言 ATS 检测 本地部署创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考