Go語(yǔ)言dolphinscheduler怎么使用

這篇文章主要介紹了Go語(yǔ)言dolphinscheduler怎么使用的相關(guān)知識(shí)，內(nèi)容詳細(xì)易懂，操作簡(jiǎn)單快捷，具有一定借鑒價(jià)值，相信大家閱讀完這篇Go語(yǔ)言dolphinscheduler怎么使用文章都會(huì)有所收獲，下面我們一起來(lái)看看吧。

自動(dòng)化

為什么需要自動(dòng)化任務(wù)處理，當(dāng)你的dolphinscheduler有幾百上千個(gè)任務(wù)，管理是非常耗時(shí)的，如果每個(gè)任務(wù)都配置郵件告警，那一有問題整天都在救火

此時(shí)就需要任務(wù)結(jié)果監(jiān)控和任務(wù)重跑來(lái)解決 失敗任務(wù)和任務(wù)自動(dòng)重跑，避免浪費(fèi)過多時(shí)間在維護(hù)dolphinscheduler任務(wù)上

使用

在調(diào)用api之前需要為用戶申請(qǐng)token,按圖操作

dolphinscheduler提供類似swagge接口UI工具

例子

該demo還是使用了http請(qǐng)求包(HttpRequest),json數(shù)據(jù)搜索包(go-jmespath)

任務(wù)結(jié)果檢查

填坑說明

• 日期處理： 使用了%20轉(zhuǎn)譯空格，使用Sprintf方法拼接字符串

• 多種數(shù)據(jù)類型: 使用interface{}來(lái)支持int,string等多種數(shù)據(jù)類型

• 數(shù)據(jù)轉(zhuǎn)換1: 將byte數(shù)據(jù)轉(zhuǎn)成json格式，方便搜索

• 數(shù)據(jù)轉(zhuǎn)換2: 將interface{}數(shù)據(jù)轉(zhuǎn)成字符串切片，方便使用

該方法可以做成周期性任務(wù)運(yùn)行，將失敗的job查出來(lái)，后續(xù)是要告警通知，還是根據(jù)job名稱查出對(duì)應(yīng)id進(jìn)行重跑任務(wù)

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是空格的轉(zhuǎn)譯
   fmt.Println(fmt.Sprintf("%v%v",today,"%2000:00:00"))
   fmt.Println(fmt.Sprintf("%v%v",tomorrow,"%2000:00:00"))
   //需要檢查的項(xiàng)目名稱
   projects := []string{"jdOrder","jdPlay"}
   //需要檢查的時(shí)間段 頁(yè)碼是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檢查狀態(tài)碼不符期望: ",resp.StatusCode())
         return
      }
      fmt.Println("resp",resp)
      //將返回?cái)?shù)據(jù)從byte轉(zhuǎn)成json格式
      body, _ := resp.Body()
      var i interface{}
      var s []string
      _ = json.Unmarshal(body, &i)
      //搜索出需要的字段對(duì)應(yīng)數(shù)據(jù)
      processInstanceNames, _ := jmespath.Search("data.totalList[*].processInstanceName", i)
      //將interface轉(zhuǎn)成[]string
      for _,v := range processInstanceNames.([]interface{}) {
         s = append(s,v.(string))
      }
      //打印出結(jié)果
      for _,v := range s {
         fmt.Println(v)
      }
   }
}

測(cè)試連接

如果上小節(jié)任務(wù)跑不成功，可以先運(yùn)行該方法，測(cè)試連接正確性

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)
}

重跑任務(wù)

重跑任務(wù)其實(shí)就是再次啟動(dòng)任務(wù),直接調(diào)用start_job既可

項(xiàng)目名稱和ID需要通過該接口獲取,這個(gè)是固定的

調(diào)用示例： 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開始狀態(tài)碼不符期望: ",resp.StatusCode())
      return
   }
}

關(guān)于“Go語(yǔ)言dolphinscheduler怎么使用”這篇文章的內(nèi)容就介紹到這里，感謝各位的閱讀！相信大家對(duì)“Go語(yǔ)言dolphinscheduler怎么使用”知識(shí)都有一定的了解，大家如果還想學(xué)習(xí)更多知識(shí)，歡迎關(guān)注億速云行業(yè)資訊頻道。