Go语言dolphinscheduler任务调度处理

 更新时间:2022年06月08日 14:45:21   作者:爱吃红薯粉  
这篇文章主要为大家介绍了Go语言dolphinscheduler任务调度处理,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪

简介

dolphinscheduler是一个可视化DAG工作流任务调度平台,在大数据领域做任务调用非常流行

提供了类似azkaban工作流调度,比azkaban更强的可视化DAG,支持大数据领域flink,spark,shell,python,java,scala,http等各种类型任务

官网传送门: https://dolphinscheduler.apache.org/zh-cn/

自动化

为什么需要自动化任务处理,当你的dolphinscheduler有几百上千个任务,管理是非常耗时的,如果每个任务都配置邮件告警,那一有问题整天都在救火

此时就需要任务结果监控和任务重跑来解决 失败任务和任务自动重跑,避免浪费过多时间在维护dolphinscheduler任务上

使用

在调用api之前需要为用户申请token,按图操作

dolphinscheduler提供类似swagge接口UI工具,访问doc地址访问

http://ip:12345/dolphinscheduler/doc.html?language=zh_CN&lang=cn

例子

该demo还是使用了http请求包(HttpRequest),json数据搜索包(go-jmespath)

任务结果检查

填坑说明

  • 日期处理: 使用了%20转译空格,使用Sprintf方法拼接字符串
  • 多种数据类型: 使用interface{}来支持int,string等多种数据类型
  • 数据转换1: 将byte数据转成json格式,方便搜索
  • 数据转换2: 将interface{}数据转成字符串切片,方便使用

该方法可以做成周期性任务运行,将失败的job查出来,后续是要告警通知,还是根据job名称查出对应id进行重跑任务

package main
import (
   "encoding/json"
   "fmt"
   "github.com/jmespath/go-jmespath"
   "github.com/kirinlabs/HttpRequest"
   "time"
)
var (
   url = "http://ip:12345/dolphinscheduler"
   token = "xxxxxxx"
   req *HttpRequest.Request
)
func init() {
   req = HttpRequest.NewRequest().Debug(true).SetTimeout(time.Second*5).
      SetHeaders(map[string]string{
         "token":token,
      })
}
func main() {
   //testConn()
   jobCheck()
}
func jobCheck()  {
   //获取日期
   today := time.Now().Format("2006-01-02")
   tomorrow := time.Now().AddDate(0, 0, +1).Format("2006-01-02")
   //拼接日期 %20是空格的转译
   fmt.Println(fmt.Sprintf("%v%v",today,"%2000:00:00"))
   fmt.Println(fmt.Sprintf("%v%v",tomorrow,"%2000:00:00"))
   //需要检查的项目名称
   projects := []string{"jdOrder","jdPlay"}
   //需要检查的时间段 页码是int类型,日期是string类型
   m := make(map[string]interface{})
   m["pageNo"] = 1
   m["pageSize"] = 22
   m["stateType"] = "FAILURE"
   m["startDate"] = fmt.Sprintf("%v%v",today,"%2000:00:00")
   m["endDate"] = fmt.Sprintf("%v%v",tomorrow,"%2000:00:00")
   for _, project := range projects {
      resp, _ := req.Get(url+"/projects/"+project+"/task-instance/list-paging",m)
      if resp.StatusCode() != 200 {
         fmt.Println("job检查状态码不符期望: ",resp.StatusCode())
         return
      }
      fmt.Println("resp",resp)
      //将返回数据从byte转成json格式
      body, _ := resp.Body()
      var i interface{}
      var s []string
      _ = json.Unmarshal(body, &i)
      //搜索出需要的字段对应数据
      processInstanceNames, _ := jmespath.Search("data.totalList[*].processInstanceName", i)
      //将interface转成[]string
      for _,v := range processInstanceNames.([]interface{}) {
         s = append(s,v.(string))
      }
      //打印出结果
      for _,v := range s {
         fmt.Println(v)
      }
   }
}

测试连接

如果上小节任务跑不成功,可以先运行该方法,测试连接正确性

func testConn() {
   resp, _ := req.Get(url + "/projects/query-project-list")
   fmt.Println("resp",resp)
   body, _ := resp.Body()
   var i interface{}
   _ = json.Unmarshal(body, &i)
   fmt.Println("i",i)
}

重跑任务

重跑任务其实就是再次启动任务,直接调用start_job既可

项目名称和ID需要通过该接口获取,这个是固定的

http://ip:12345/dolphinscheduler/projects/monitor/process/list-paging

调用示例: startJob("ads_jd_order",678)

func startJob(projectName string,projectId int)  {
   m := make(map[string]interface{})
   m["failureStrategy"] = "CONTINUE"
   m["warningGroupId"] = 0
   m["warningType"] = "NONE"
   m["runMode"] = "RUN_MODE_SERIAL"
   m["processInstancePriority"] = "MEDIUM"
   m["workerGroup"] = "default"
   m["processDefinitionId"] = projectId
   resp, _ := req.JSON().Post(url+"projects/" + projectName+"/executors/start-process-instance",m)
   if resp.StatusCode() != 200 {
      fmt.Println("job开始状态码不符期望: ",resp.StatusCode())
      return
   }
}

小结

dolphinscheduler api调用有文档,不太复杂,但网上资料较少,需要自行摸索,以上就是Go语言dolphinscheduler任务调度处理的详细内容,更多关于Go语言dolphinscheduler任务调度的资料请关注脚本之家其它相关文章!

相关文章

  • go语言的变量定义示例详解

    go语言的变量定义示例详解

    这篇文章主要为大家介绍了go语言的变量定义示例详解,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
    2022-12-12
  • GoLang并发机制探究goroutine原理详细讲解

    GoLang并发机制探究goroutine原理详细讲解

    goroutine是Go语言提供的语言级别的轻量级线程,在我们需要使用并发时,我们只需要通过 go 关键字来开启 goroutine 即可。这篇文章主要介绍了GoLang并发机制goroutine原理,感兴趣的可以了解一下
    2022-12-12
  • Go项目分层下的最佳error处理方式分享

    Go项目分层下的最佳error处理方式分享

    这篇文章主要来和大家一起探讨 Go 项目分层下的最佳 error 处理方式,准备好了吗?准备一杯你最喜欢的饮料或茶,随着本文一探究竟吧
    2023-06-06
  • GOLANG使用Context管理关联goroutine的方法

    GOLANG使用Context管理关联goroutine的方法

    这篇文章主要介绍了GOLANG使用Context管理关联goroutine的方法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2019-01-01
  • Goland的设置与配置全过程

    Goland的设置与配置全过程

    这篇文章主要介绍了Goland的设置与配置全过程,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2023-02-02
  • 利用Go语言搭建WebSocket服务端方法示例

    利用Go语言搭建WebSocket服务端方法示例

    这篇文章主要给大家介绍了利用Go语言搭建WebSocket服务端方法,文中通过示例代码介绍的非常详细,需要的朋友们可以参考借鉴,下面来一起看看吧。
    2017-04-04
  • GoLang桥接模式的实现示例

    GoLang桥接模式的实现示例

    桥接模式是一种结构型设计模式,通过桥接模式可以将抽象部分和它的实现部分分离,本文主要介绍了GoLang桥接模式,文中通过示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2022-01-01
  • 浅析Go语言bitset的实现原理

    浅析Go语言bitset的实现原理

    bitset包是一个将非负整数映射到布尔值的位的集合,这篇文章主要通过开源包bitset来为大家分析一下位集合的设计和实现,感兴趣的可以学习一下
    2023-08-08
  • Golang学习之无类型常量详解

    Golang学习之无类型常量详解

    对于无类型常量,可能大家是第一次听说,但我们每天都在用,每天都有无数潜在的坑被埋下。本文就来和大家聊聊它的相关注意事项吧,希望对大家有所帮助
    2023-03-03
  • 详解go-zero如何使用validator进行参数校验

    详解go-zero如何使用validator进行参数校验

    这篇文章主要介绍了如何使用validator库做参数校验的一些十分实用的使用技巧,包括翻译校验错误提示信息、自定义提示信息的字段名称、自定义校验方法等,感兴趣的可以了解下
    2024-01-01

最新评论