这类工具最值得先看的不是功能列表而是能不能在普通环境里稳定跑起来。DolphinDB 的批处理作业说白了就是帮你把一堆定时或按需的数据计算任务管起来不用你手动一个个去点。它适合需要定期跑数据清洗、报表生成、模型训练结果更新的数据分析师和开发。最关键的能力是能把任务编排、依赖管理和执行监控这些事从你写的业务代码里抽离出来让脚本更干净也让任务运行更可控。我建议先从最小样例开始。很多人在接触这类功能时容易把“批处理”想得太复杂要么一上来就想调度几百个任务要么觉得必须有个复杂的界面。其实核心就两步第一把你的计算逻辑包装成一个可执行的任务单元第二告诉系统这个任务什么时候跑、依赖谁、结果存哪。下面按实际落地顺序拆一遍。1. 先搞清楚 DolphinDB 里“批处理作业”指的是什么很多人听到“批处理”会直接想到 Hadoop 或者 Spark 那种大数据计算框架但在 DolphinDB 的语境里它更接近一个任务调度与执行管理器。它的核心不是做分布式计算虽然 DolphinDB 本身支持分布式而是帮你把已经写好的脚本按照你设定的计划或者触发条件自动、可靠地执行起来。1.1 和手动执行脚本的核心区别如果你现在每天手动登录服务器然后执行一个.dos脚本文件来更新日报这就是手动模式。这种方式有几个明显的问题依赖人工你必须在特定时间点操作容易忘记或延误。难以监控脚本是成功还是失败失败了报什么错你需要自己去查日志没有集中视图。没有依赖管理如果任务 B 需要任务 A 的输出结果你得等 A 跑完手动确认再跑 B。缺乏容错任务中途出错通常就停在那里需要人工介入。DolphinDB 的批处理作业功能就是为了把“手动”变成“自动”把“散装”变成“流水线”。它提供了一个框架让你能定义任务Job、设置调度计划Schedule、管理任务之间的依赖关系并查看执行历史和日志。1.2 关键组件Job 和 Schedule这是两个最核心的概念必须分清楚Job作业这是一个具体的计算任务单元。它就是你写的一个脚本比如“计算今日股票收益率的十分位数”。你需要把这个脚本提交给 DolphinDB 的批处理系统它就成了一个待管理的 Job。Schedule调度计划这是告诉系统“什么时候、以什么频率”去执行 Job 的规则。比如每天下午 4 点收盘后执行或者每小时执行一次。一个 Job 可以关联多个 Schedule一个 Schedule 也可以触发多个 Job通常通过依赖关系串联。理解了这个你就知道配置时主要在做两件事定义任务内容定义触发规则。2. 环境准备与第一个“Hello World”批处理作业不要一上来就配置生产环境的复杂依赖。我更建议在测试环境甚至单机版的 DolphinDB 里先把整个流程跑通。这里最容易忽略的是路径和权限。2.1 基础环境确认首先确保你的 DolphinDB 服务已经启动并能正常连接。无论是通过 DolphinDB GUI、VS Code 插件还是dolphindb命令行工具你能连上就行。 接着确认你有权限创建和提交作业。通常管理员账号默认都有权限。你可以通过以下脚本快速检查批处理相关函数是否可用# 这是一个在 DolphinDB 脚本中的检查假设你在 Python 客户端用 run 函数执行 # 实际上你是在 DolphinDB GUI 或脚本文件中写 DolphinDB 的脚本语言 # 以下为 DolphinDB 脚本示例 try { scheduleJob(); print(批处理调度功能可用。); } catch (ex) { print(当前环境可能不支持或未启用批处理功能错误信息, ex); }如果报错可能需要检查 DolphinDB 的版本是否支持或者是否以开启了相应模块社区版通常包含。2.2 创建并提交你的第一个 Job假设我们有一个最简单的任务向一个指定表里插入一条带时间戳的日志记录表示任务开始。我们在 DolphinDB 里先创建这个目标表// 在 DolphinDB 脚本中执行 // 创建一个简单的日志表如果不存在的话 if(existsTable(dfs://demoBatch, jobLog) false) { db database(dfs://demoBatch, VALUE, 2024.01.01..2024.12.31) tb table(1:0, jobIdjobNamestartTimestatus, [SYMBOL, SYMBOL, TIMESTAMP, SYMBOL]) pt db.createPartitionedTable(tb, jobLog, startTime) }现在编写任务脚本myFirstJob.dos// myFirstJob.dos 内容 jobId test_001; jobName Daily_Summary; startTime now(); status STARTED; // 获取或创建日志表句柄 login(admin, 123456) // 按实际环境修改 db database(dfs://demoBatch) pt loadTable(db, jobLog) // 插入日志 pt.append!(table(jobId, jobName, startTime, status as jobIdjobNamestartTimestatus)); print(Job [, jobId, ] started at: , startTime);接下来提交这个脚本作为一个批处理 Job。这是关键一步不是直接执行脚本而是把它“注册”到调度系统。使用scheduleJob函数// 提交作业这里先不设置调度仅提交 scheduleJob(jobIdtest_001, jobDesc我的第一个测试作业, jobScriptmyFirstJob.dos, scheduleTime[], recurringfalse, priority0, parallelfalse)解释一下参数jobId: 作业的唯一标识很重要后续查询、管理都靠它。jobDesc: 作业描述方便人阅读。jobScript: 作业脚本的文件路径或脚本内容本身。这里用的是文件路径。scheduleTime: 调度时间列表。设为空数组[]表示不自动调度需要手动触发。recurring: 是否循环执行。false表示只执行一次。priority: 优先级。parallel: 是否允许并行执行如果前一个实例还没跑完。执行完scheduleJob你的第一个 Job 就定义好了但它还不会自动运行因为我们没给调度计划。2.3 手动触发与立即执行测试对于刚定义的 Job最直接的测试方法是手动触发一次看看它能不能跑通以及输出和日志是否符合预期。// 手动运行指定的 Job runJob(test_001);执行后去检查dfs://demoBatch数据库下的jobLog表应该能看到一条记录。同时在 DolphinDB 的日志文件或 GUI 的“作业”查看界面里能看到print语句输出的信息。这是非常重要的验证环节确保你的脚本在批处理上下文中能独立、正确地运行。很多错误源于脚本内使用了未定义的变量或依赖了交互式会话中的临时对象。3. 给作业加上调度计划从单次到周期执行单次手动执行没问题后就可以给它加上“闹钟”了。这就是配置Schedule。3.1 单次定时执行假设我们需要在今天的下午 3 点整执行一次test_001这个 Job。我们需要修改或重新提交这个 Job这次指定scheduleTime。// 假设当前日期是 2024.05.20我们设定今天 15:00:00 执行 targetTime timestamp(2024.05.20T15:00:00.000); // 注意如果 jobId test_001 已存在需要先删除旧定义或者使用 updateJob 函数如果版本支持。 // 这里演示先删除再创建。生产环境请谨慎操作。 cancelJob(test_001); // 取消已有调度如果存在 // 重新提交并指定调度时间 scheduleJob(jobIdtest_001, jobDesc下午三点执行的测试, jobScriptmyFirstJob.dos, scheduleTimetargetTime, recurringfalse, priority0, parallelfalse)提交后系统会在2024.05.20T15:00:00自动触发这个 Job 的执行。你可以通过getScheduledJobs函数查看所有已调度的作业。3.2 循环执行每日、每周、每月更常见的场景是周期性任务比如每天收盘后运行。这就需要设置recurringtrue并指定循环规则。DolphinDB 通过scheduleTime列表和recurring参数配合实现。 例如设置每天下午 4 点执行// 定义每天 16:00 执行 dailyTime 16:00:00.000; // 取消旧作业如果存在 cancelJob(test_001); // 提交每日作业 scheduleJob(jobIdtest_001, jobDesc每日收盘作业, jobScriptmyFirstJob.dos, scheduleTimedailyTime, recurringtrue, priority0, parallelfalse)这里的scheduleTime是一个时间TIME类型列表。系统会从下一个匹配该时间的点开始每日重复。例如你在今天下午 5 点提交那么第一次执行将在明天下午 4 点。对于更复杂的周期比如每周一上午 9 点scheduleTime可以包含多个时间点并结合daysOfWeek参数具体请查阅对应版本手册这里不展开因为“上篇”聚焦基础。3.3 查看作业状态与历史作业提交后你怎么知道它成功运行了还是失败了不要等到业务出问题才去查。DolphinDB 提供了查询函数。查看待调度作业getScheduledJobs()会列出所有已定义并等待触发的作业。查看作业执行历史getJobHistory()或getJobHistory(jobId)可以查看作业的运行记录包括开始时间、结束时间、状态成功/失败和错误信息如果有。查看最近作业状态getRecentJobs()查看最近一段时间内的作业执行情况。养成习惯在提交或修改作业后用getScheduledJobs确认一下调度时间是否正确。在预期执行时间点过后用getJobHistory检查是否成功执行。这是判断批处理系统是否正常工作的直接依据。4. 从单任务到任务链理解依赖与执行顺序单个任务自动化只是第一步。真实场景中任务往往有前后依赖B 任务需要 A 任务产出的数据C 任务需要在 A 和 B 都成功后才能开始。DolphinDB 批处理支持这种依赖关系。4.1 通过“时间差”实现隐式依赖最简单但不推荐的依赖方式是靠调度时间错开。比如设置任务 A 在 16:00 跑任务 B 在 16:05 跑假设 A 任务 5 分钟内能跑完。这种方式非常脆弱一旦 A 任务执行超时或失败B 任务依然会准时启动导致错误。4.2 使用runJob在脚本中显式触发下游任务更可靠的方式是在任务 A 的脚本末尾成功执行后主动调用runJob来触发任务 B。这样形成了直接的链式调用。// 任务A脚本 (jobA.dos) // ... 执行A的核心逻辑 ... print(“Task A finished successfully.”); // 显式触发任务B try { runJob(job_B); print(“Triggered Job B.”); } catch (ex) { print(“Failed to trigger Job B: “, ex); }这种方式将依赖逻辑写在了业务脚本里优点是直接、清晰。缺点是耦合度高如果任务链变更需要修改多个脚本。4.3 使用批处理系统的依赖配置如depends更优雅的方式是利用批处理系统自身的依赖管理功能。在某些版本的 DolphinDB 或通过特定函数/界面你可以定义 Job 之间的依赖关系图。例如提交 Job B 时指定其depends[job_A]。这样系统会在 Job A 成功完成后自动触发 Job B而无需在 A 的脚本里写触发代码。 这是更“批处理”的做法将调度逻辑和业务逻辑分离。你需要查阅你所使用版本的 DolphinDB 手册确认scheduleJob或相关函数是否支持depends参数或者是否有专门的addDependency 函数。在落地时我建议先从“脚本内显式触发”开始因为它最简单直观能快速验证任务链逻辑。待核心流程跑通后再研究如何迁移到系统的依赖配置上以获得更好的可维护性和可视化。5. 实操中的常见问题与排查顺序当你按照上述步骤操作时可能会遇到作业没按时跑、或者跑失败了的情况。不要急着修改脚本先按顺序排查。5.1 作业根本没触发检查点1调度时间是否正确。使用getScheduledJobs()确认你定义的scheduleTime是否符合预期。注意时区问题DolphinDB 默认使用服务器本地时区。检查点2作业是否被禁用或取消。同样在getScheduledJobs()的结果中查看作业状态是否为ACTIVE。检查点3DolphinDB 服务是否在调度时间点正常运行。检查服务器日志看是否有异常重启。5.2 作业触发但执行失败检查点1查看作业历史详情。使用getJobHistory(jobId)找到失败的那次执行记录查看errorMsg字段。这是最直接的错误信息。检查点2检查脚本中的路径和权限。批处理作业运行时可能使用特定的用户或上下文其工作目录、数据库访问权限可能与你在 GUI 中交互时不同。确保脚本内使用的数据库路径、文件路径都是绝对路径或者相对于批处理作业运行环境的正确路径。检查点3检查脚本依赖的外部变量或函数。确保脚本中引用的所有共享变量、自定义函数在作业执行时都是已定义的。最好在脚本开头显式地include必要的模块或脚本文件。检查点4资源是否充足。如果作业涉及大量数据计算可能因内存不足、磁盘空间不够而失败。查看系统资源监控。5.3 作业执行成功但结果不对检查点1确认输入数据。作业是否处理了正确的数据分区或时间范围特别是基于时间调度的作业要检查脚本中用于过滤数据的时间变量如today()在批处理上下文中的值是否符合预期。检查点2验证输出。直接查询作业输出的目标表或文件检查数据量、数据内容是否正常。与手动执行脚本的结果进行对比。检查点3检查并发冲突。如果设置了paralleltrue或者多个作业同时操作同一张表可能存在读写冲突。考虑是否需要加锁或调整调度时间错开。5.4 性能与稳定性建议日志是生命线在作业脚本中关键步骤加入print或使用writeLog函数输出详细日志。这能让你在出问题时快速定位阶段。设置超时与重试对于可能不稳定的任务如依赖外部网络在脚本内实现简单的重试机制或者研究 DolphinDB 是否支持作业级别的超时和重试配置。从小批量开始不要一开始就用全量数据测试调度。先用一天、一小时的数据跑通整个流程确认无误后再扩展到更大规模。监控与报警将getJobHistory的查询与监控系统结合对连续失败或长时间运行的作业设置报警。我个人更建议先把单任务跑稳再考虑任务链和复杂调度。这个方案真正落地时最该盯住的不是功能列表而是脚本的独立性、路径权限和日志可查性。踩过几次之后我发现很多问题不是批处理系统能力不够而是提交的脚本本身在批处理环境下无法独立运行。所以提交前多用runJob手动触发几次确保它在“无人值守”模式下也能工作正常。
