Go語(yǔ)言dolphinscheduler任務(wù)調(diào)度處理
簡(jiǎn)介
dolphinscheduler是一個(gè)可視化DAG工作流任務(wù)調(diào)度平臺(tái),在大數(shù)據(jù)領(lǐng)域做任務(wù)調(diào)用非常流行
提供了類(lèi)似azkaban工作流調(diào)度,比azkaban更強(qiáng)的可視化DAG,支持大數(shù)據(jù)領(lǐng)域flink,spark,shell,python,java,scala,http等各種類(lèi)型任務(wù)
官網(wǎng)傳送門(mén): https://dolphinscheduler.apache.org/zh-cn/
自動(dòng)化
為什么需要自動(dòng)化任務(wù)處理,當(dāng)你的dolphinscheduler有幾百上千個(gè)任務(wù),管理是非常耗時(shí)的,如果每個(gè)任務(wù)都配置郵件告警,那一有問(wèn)題整天都在救火
此時(shí)就需要任務(wù)結(jié)果監(jiān)控和任務(wù)重跑來(lái)解決 失敗任務(wù)和任務(wù)自動(dòng)重跑,避免浪費(fèi)過(guò)多時(shí)間在維護(hù)dolphinscheduler任務(wù)上
使用
在調(diào)用api之前需要為用戶(hù)申請(qǐng)token,按圖操作
dolphinscheduler提供類(lèi)似swagge接口UI工具,訪問(wèn)doc地址訪問(wèn)
http://ip:12345/dolphinscheduler/doc.html?language=zh_CN&lang=cn
例子
該demo還是使用了http請(qǐng)求包(HttpRequest),json數(shù)據(jù)搜索包(go-jmespath)
任務(wù)結(jié)果檢查
填坑說(shuō)明
- 日期處理: 使用了%20轉(zhuǎn)譯空格,使用Sprintf方法拼接字符串
- 多種數(shù)據(jù)類(lèi)型: 使用interface{}來(lái)支持int,string等多種數(shù)據(jù)類(lèi)型
- 數(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名稱(chēng)查出對(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)目名稱(chēng) projects := []string{"jdOrder","jdPlay"} //需要檢查的時(shí)間段 頁(yè)碼是int類(lèi)型,日期是string類(lèi)型 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)目名稱(chēng)和ID需要通過(guò)該接口獲取,這個(gè)是固定的
http://ip:12345/dolphinscheduler/projects/monitor/process/list-paging
調(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開(kāi)始狀態(tài)碼不符期望: ",resp.StatusCode()) return } }
小結(jié)
dolphinscheduler api調(diào)用有文檔,不太復(fù)雜,但網(wǎng)上資料較少,需要自行摸索,以上就是Go語(yǔ)言dolphinscheduler任務(wù)調(diào)度處理的詳細(xì)內(nèi)容,更多關(guān)于Go語(yǔ)言dolphinscheduler任務(wù)調(diào)度的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
GoLang并發(fā)機(jī)制探究goroutine原理詳細(xì)講解
goroutine是Go語(yǔ)言提供的語(yǔ)言級(jí)別的輕量級(jí)線程,在我們需要使用并發(fā)時(shí),我們只需要通過(guò) go 關(guān)鍵字來(lái)開(kāi)啟 goroutine 即可。這篇文章主要介紹了GoLang并發(fā)機(jī)制goroutine原理,感興趣的可以了解一下2022-12-12GOLANG使用Context管理關(guān)聯(lián)goroutine的方法
這篇文章主要介紹了GOLANG使用Context管理關(guān)聯(lián)goroutine的方法,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2019-01-01利用Go語(yǔ)言搭建WebSocket服務(wù)端方法示例
這篇文章主要給大家介紹了利用Go語(yǔ)言搭建WebSocket服務(wù)端方法,文中通過(guò)示例代碼介紹的非常詳細(xì),需要的朋友們可以參考借鑒,下面來(lái)一起看看吧。2017-04-04淺析Go語(yǔ)言bitset的實(shí)現(xiàn)原理
bitset包是一個(gè)將非負(fù)整數(shù)映射到布爾值的位的集合,這篇文章主要通過(guò)開(kāi)源包bitset來(lái)為大家分析一下位集合的設(shè)計(jì)和實(shí)現(xiàn),感興趣的可以學(xué)習(xí)一下2023-08-08Golang學(xué)習(xí)之無(wú)類(lèi)型常量詳解
對(duì)于無(wú)類(lèi)型常量,可能大家是第一次聽(tīng)說(shuō),但我們每天都在用,每天都有無(wú)數(shù)潛在的坑被埋下。本文就來(lái)和大家聊聊它的相關(guān)注意事項(xiàng)吧,希望對(duì)大家有所幫助2023-03-03詳解go-zero如何使用validator進(jìn)行參數(shù)校驗(yàn)
這篇文章主要介紹了如何使用validator庫(kù)做參數(shù)校驗(yàn)的一些十分實(shí)用的使用技巧,包括翻譯校驗(yàn)錯(cuò)誤提示信息、自定義提示信息的字段名稱(chēng)、自定義校驗(yàn)方法等,感興趣的可以了解下2024-01-01