分支节点实现特定时间执行任务

分支节点产生背景

Cron表达式无法实现一个节点需要每个月的最后一天执行的场景，分支节点产生后，您可以套用switch-case编程模型实现该需求。详情请参见分支节点。

分支节点与其它控制节点

在数据开发页面，您可以看到当前版本的DataWorks支持的赋值节点、分支节点和归并节点等控制节点。

• ASN：一个赋值节点，用于对比较复杂的情况进行计算，以便选择分支节点的条件。

• X和Y：分支节点，处于赋值节点ASN下游，根据赋值节点的输出进行分支的选择。如图中绿色线条所示，X节点选择了左边的分支，Y节点选择了左边两个分支：

  • A和C节点由于在X和Y节点被选择的输出下游，因此正常执行。

  • B节点虽然在Y节点被选择的分支下游，但由于X节点未选择该输出，因此B节点被置为空跑。

  • E节点由于未被Y节点选中，因此即使有一个普通的Z节点上游，也同样被置为了空跑。

  • G节点由于上游E节点空跑，因此即使C/F都正常执行，G节点同样空跑。

  • 空跑属性什么情况下才能不再向下传递？

    JOIN节点是一个归并节点，它的特殊功能就是停止空跑属性的传递，可以看到由于D节点处于JOIN节点下游，因此B节点的空跑属性被阻断了，D节点可以开始正常跑了。

您可以通过利用分支节点配合其它控制节点，满足某个节点只有每个月最后一天运行的需求场景。

使用分支节点

1. 定义任务依赖。

   1. 根节点赋值节点通过定时时间SKYNET_CYCTIME来计算当前是否为本月的最后一天。如果是，则输出1。如果不是，则输出0。该输出会被DataWorks捕获，传递给下游。

   2. 分支节点通过赋值节点的输出来定义分支。

   3. 两个Shell节点挂在分支节点下，分别执行不同的分支逻辑。

2. 定义赋值节点。

   新建赋值节点时，会自带一个outputs，赋值节点支持编写SQL、SHELL和Python三种语言。

   • 对于SQL类型，DataWorks捕获最后一条SELECT语句作为outputs的值。

   • 对Shell和Python类型，DataWorks捕获最后一行标准输出作为outputs的值。

   本文采用Python类型作为赋值节点的代码，调度属性和代码设置如下。

   • 代码设置

     
     import os
     import time
     
     dueTimeStr = os.environ['SKYNET_CYCTIME']  # 20190104000200
     dueTime = time.strptime(dueTimeStr[:8], "%Y%m%d")
     dueMonth = time.strftime("%m", dueTime)
     
     nextDueTimeStamp = int(time.mktime(dueTime))+3600*24
     nextDueTime = time.localtime(nextDueTimeStamp)
     nextDueMonth = time.strftime("%m", nextDueTime)
     
     print ('Current month: %s, next month: %s') % (dueMonth, nextDueMonth)
     
     if nextDueMonth == dueMonth:
         print 0
     else:
         print 1
     

   • 调度属性配置

     赋值节点的本节点输出参数中默认包含参数 outputs，类型为变量，取值为 ${outputs}，该值由运行时决定。outputs 是赋值节点的核心能力，用于将赋值节点的运行结果传递给下游分支节点。

3. 定义分支。

   分支节点可以用Python语法的表达式定义条件，每个条件会绑定一个输出。当满足条件时，该输出的下游节点会被执行，而其它的节点会被置为空跑。

   • 调度配置

     在分支节点的调度配置中，节点上下文的输入参数 isLast 取值来源为赋值节点的输出（最后一天 isLast 为 1，其他情况 isLast 为 0）。输出参数 outputs 取值为 ${outputs}，由运行时决定。本节点的输出绑定了两个下游节点：只有最后一天执行和除了最后一天之外运行，分支走向根据 isLast 的值决定。

   • 分支配置

     在分支逻辑定义中，根据 isLast 变量定义两个分支：

     • 条件 ${isLast}==0，关联到节点输出 autotest.last_day_cond.not_last，表示当天不是最后一天时使用该分支

     • 条件 ${isLast}==1，关联到节点输出 autotest.last_day_cond.is_last，表示当前是最后一天时使用该分支

   • 调度配置生成条件绑定的输出

     在本节点的输出中，系统会为每个分支条件默认添加与条件同名的输出，例如 autotest.last_day_cond.is_last 和 autotest.last_day_cond.not_last。输出表中可查看各输出对应的下游节点名称、下游节点 ID、责任人及来源信息。

4. 将执行任务节点挂在不同分支下。

   分支节点有3个输出，任意选择一个输出当作输入即可。由于现在分支节点的输出关联了条件，所以要慎重选择。

   • 每月最后一天执行的节点依赖

     在只有最后一天执行节点的调度依赖面板中，将自动解析设置为是，在依赖的上游节点中手动添加父节点输出名称为 autotest.last_day_cond.is_last 的依赖（对应节点：分支_根据最后一天决定下游执行），即将该节点挂在每月最后一天的分支下面。

   • 每月其它时间执行的节点依赖

     在节点除了最后一天之外运行的调度依赖配置中，手动添加父节点输出名称为 autotest.last_day_cond.not_last，即将该节点挂在分支节点分支_根据最后一天决定下游执行的"非最后一天"分支下面，使其仅在非月末时被调度执行。

5. 验证结果

   完成上述所有配置后，您可以提交并发布任务。完成发布后，可以执行补数据来测试效果：业务日期选择2018-12-30和2018-12-31，即定时时间分别为2018-12-31和2019-01-01，则第一批补数据会触发最后一天的逻辑，第二批触发非最后一天的逻辑。两者的区别如下所示。

   业务日期为2018-12-30（即定时时间为2018-12-31）

   • 分支节点分支选择结果

     
     2019-01-04 23:26:19.867   INFO  - The following profiles are active: dev
     2019-01-04 23:26:21.239   INFO  - Started ControllerWrapper in 2.072 seconds (JVM runni...
     python output: False
     2019-01-04 23:26:21.419   INFO  - ...
     2019-01-04 23:26:22.424   INFO  - not meet the condition! condition:1==0
     python output: True
     2019-01-04 23:26:22.435   INFO  - ...
     2019-01-04 23:26:23.436   INFO  - meet the condition. condition:1==1
     2019-01-04 23:26:23.437   INFO  - ===>Output Result: autotest.last_day_cond.is_last
     2019-01-04 23:26:24.070   INFO  - cost Time: 2
     2019-01-04 23:26:24.071   INFO  - job finished!
     

   • 节点（最后一天执行）正常执行

   • 节点（除了最后一天之外运行）被置为空跑

   业务日期为2018-12-31（即定时时间为2019-01-01）

   • 分支节点分支选择结果

     分支节点的运行日志显示条件判断过程：条件 0==0 满足（输出 True），条件 0==1 不满足（输出 False），最终输出结果为 ===>Output Result: autotest.last_day_cond.not_last，即选择了非最后一天的分支。

   • 节点（最后一天执行）被置为空跑

   • 节点（除了最后一天之外运行）正常执行

总结