定时作业

以下例子调用了自定义函数 getMaxTemperature，在每天0点计算前一天某个设备温度指标的最大值。

def getMaxTemperature(deviceID){
    maxTemp=exec max(temperature) from loadTable("dfs://dolphindb","sensor")
            where ID=deviceID ,ts between (today()-1).datetime():(today().datetime()-1)
    return  maxTemp
}
scheduleJob(`testJob, "getMaxTemperature", getMaxTemperature{1}, 00:00m, today(), today()+30, 'D');

注意：自定义函数 getMaxTemperature 的参数是设备编号，部分应用 getMaxTemperature{1} 代表给设备编号赋值1。

以下例子中，在2020年每个月1号的0点使用 run 函数执行脚本文件 monthlyJob.dos：

scheduleJob(`monthlyJob, "Monthly Job 1", run{"/home/DolphinDB/script/monthlyJob.dos"}, 00:00m, 2020.01.01, 2020.12.31, 'M', 1);

注意：指定脚本文件 monthlyJob.dos 的完整路径作为部分应用 run 函数的参数。

在以下例子中，每周日的 1 点会定时执行删除日志文件的操作系统命令。作业函数为 shell 函数，并指定命令 rm /home/DolphinDB/server/dolphindb.log 作为参数。

scheduleJob(`weeklyjob, "rm log", shell{"rm /home/DolphinDB/server/dolphindb.log"}, 01:00m, 2020.01.01, 2021.12.31, `W`, 0);

在实际应用中，更常用的做法是从数据库中取出数据作为参数传入作业函数，计算后将结果存入数据库。示例4中的自定义函数 computeK，从分布式数据库表 trades 中取出行情数据，计算后将结果存入分布式数据库表 OHLC 中。

以下例子展示了如何在每周一到每周五的15点，计算分钟级的 K 线：

def computeK(){
	barMinutes = 7
	sessionsStart=09:30:00.000 13:00:00.000
	OHLC =  select first(price) as open, max(price) as high, min(price) as low,last(price) as close, sum(volume) as volume 
		from loadTable("dfs://stock","trades")
		where time > today() and time < now()
		group by symbol, dailyAlignedBar(timestamp, sessionsStart, barMinutes*60*1000) as barStart
	append!(loadTable("dfs://stock","OHLC"),OHLC)
}
scheduleJob(`kJob, "7 Minutes", computeK, 15:00m, 2020.01.01, 2021.12.31, 'W', [1,2,3,4,5]);

其中，参数 onComplete 是一个有 4 个参数的回调函数，当定时作业执行完毕（包括有异常的情况）后，会执行该函数。

以下脚本运行前需安装 HttpClient 插件。

def sendEmail(jobId, jobDesc, success, result){
desc = "jobId=" + jobId + " jobDesc=" + jobDesc
if(success){
desc += " successful " + result
res = httpClient::sendEmail('patrick.mahomes@dolphindb.com','password','andy.reid@dolphindb.com','This is a subject',desc)
}
else{
desc += " with error: " + result
res = httpClient::sendEmail('patrick.mahomes@dolphindb.com','password','andy.reid@dolphindb.com','This is a subject',desc)
}
}
scheduleJob(jobId=`PnL, jobDesc="Calculate Profit & Loss", jobFunc=run{"PnL.dos"}, scheduleTime=[12:00m, 02

首先定义一个定时作业：

def foo(){
	print "test scheduled job at"+ now()
	return now()
}
scheduleJob(`testJob, "foo", foo, 17:00m+0..2*30, today(), today(), 'D');

运行 getRecentJobs() 得到示例 5 的作业信息。

jobId	            jobDesc	startTime	            endTime
------              ------- ----------------------- ----------------------
testJob	            foo 	2020.02.14T17:00:23.636	2020.02.14T17:00:23.639
testJob20200214	    foo 	2020.02.14T17:30:23.908	2020.02.14T17:30:23.910
testJob20200214000  foo 	2020.02.14T18:00:23.148	2020.02.14T18:00:26.749

第一次执行的作业 ID 是“testJob”，第二次执行的作业 ID 是“testJob20200214”，……，每次执行的作业 ID 不同。根据相应的作业ID，我们可用函数 getJobMessage 和 getJobReturn 查看第3次作业的执行情况。

>getJobMessage(`testJob20200214000);
2020-02-14 18:00:23.148629 Start the job [testJob20200214000]: foo
2020-02-14 18:00:23.148721 test the scheduled job at 2020.02.14T18:00:23.148
2020-02-14 18:00:26.749111 The job is done.

>getJobReturn(`testJob20200214000);
2020.02.14T18:00:23.148

本例中，没有访问 DFS 分布式表的权限的用户 guestUser1 将创建一个定时作业。

def foo1(){
	print "Test scheduled job "+ now()
	cnt=exec count(*) from loadTable("dfs://FuturesContract","tb")
	print "The count of table is "+cnt
	return cnt
}
login("guestUser1","123456")
scheduleJob(`guestGetDfsjob, "dfs read", foo1, [12:00m, 21:03m, 21:45m], 2020.01.01, 2021.12.31, "D");

作业执行后，使用函数 getJobMessage(`guestGetDfsjob) 进行查询。结果如下所示，定时作业没有权限读取分布式数据库：

2020-02-14 21:03:23.193039 Start the job [guestGetDfsjob]: dfs read
2020-02-14 21:03:23.193092 Test the scheduled job at 2020.02.14T21:03:23.193
2020-02-14 21:03:23.194914 Not granted to read table dfs://FuturesContract/tb

因此，若定时作业要访问集群中的某个分布式表，需要先以管理员 (admin) 或其他授权用户身份登录。具体操作可以通过函数 login 完成。

注意：

从上述日志中可以发现，访问分布式表后的语句没有被执行，即作业执行过程中若遇到错误，执行就会中断。为防止出现异常而停止执行后续脚本，可使用 try-catch 语句俘获异常。代码运行中可以使用函数 print 打印运行信息，输出结果记录在日志文件 <jobId>.msg 中。

对经过编译的函数进行序列化，只序列化函数名称和模块名称。反序列化时，会在系统中搜索这些模块及函数，若搜索不到则失败。

若已创建的定时作业中使用了插件函数，必须在反序列化前进行预先加载，否则系统启动失败。系统与定时作业的相关组件资源的初始化顺序依次是：系统级初始化脚本（dolphindb.dos），函数视图（function view），用户级启动脚本（startup.dos）和定时作业。示例7进行代码演示。

脚本函数会序列化函数参数以及函数定义的每一个语句。语句中若又包含了依赖的脚本函数，也会序列化这些依赖函数的定义。

创建定时作业后，若这些脚本函数被删除或被修改了，或它依赖的脚本函数被修改，不影响定时作业运行。若希望定时作业按新的函数执行，就需要先删除定时作业、然后重新创建定时作业，否则会运行已序列化的函数。其中要注意关联的函数也需要重新定义。下面举例说明：

下面进行举例说明：

foo()

然后在 DolphinDB GUI 中执行下列脚本：

def foo(){
	print ("Hello world!")
}
run "/home/user/testjob.dos"

结果显示可以正常执行：

2020.02.14 13:47:00.992: executing code (line 104-108)...
Hello world!

再创建定时作业使用函数 run 运行该脚本文件，代码如下所示：

scheduleJob(`dailyfoofile1, "Daily Job 1", run {"/home/user/testjob.dos"}, 16:14m, 2020.01.01, 2020.12.31, `D`);

直接运行作业，将会发生如下异常：

Exception was raised when running the script [/home/user/testjob.dos]:Syntax Error: [line #3] Cannot recognize the token foo

异常原因：函数 foo 的定义和定时作业的执行不在同一个会话 (session) 中，作业执行时找不到函数定义。

把函数 foo 的定义放到脚本文件中，修改 testjob.dos 文件内容如下：

def foo(){
	print ("Hello world!")
}
foo()

再重新创建定时作业运行这个脚本文件，即可顺利完成。