首頁 > 軟體

Go語言dolphinscheduler任務排程處理

2022-06-08 22:02:10

簡介

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任務排程的資料請關注it145.com其它相關文章!


IT145.com E-mail:sddin#qq.com