本文使用的GO是1.17.2
Doris 0.15.0 release版
Doris的数据导入有各种语言的版本，但是GO语言版本的基本见不到，简单学了一下，写了一个简单的Stream Load入库的示例，仅供参考
示例中使用的表结构：
CREATE TABLE IF NOT EXISTS user_info ( user_id LARGEINT NOT NULL COMMENT "用户id", username varchar(50) NOT NULL COMMENT "用户名", city VARCHAR(20) COMMENT "用户所在城市", age SMALLINT COMMENT "用户年龄", sex TINYINT COMMENT "用户性别", phone LARGEINT COMMENT "电话", address VARCHAR(500) COMMENT "地址", register_time datetime COMMENT "用户注册时间" ) Unique KEY(user_id, username) DISTRIBUTED BY HASH(user_id) BUCKETS 3 PROPERTIES ( "replication_num" = "3" );
下面是GO的示例代码，其中支持从文件导入，从内存数据导入，同时提供了获取BE节点列表的方法，你在导入的时候可以从这里随机获取一个BE节点IP及端口，直连BE进行导入
package main import ( "container/list" "encoding/base64" "encoding/json" "fmt" "github.com/gofrs/uuid" "io/ioutil" "log" "net/http" "strconv" "strings" ) type StreamLoad struct { url string dbName string tableName string data string userName string password string } //实现Doris用户认证信息 func auth(load StreamLoad) string { s := load.userName + ":" + load.password b := []byte(s) sEnc := base64.StdEncoding.EncodeToString(b) fmt.Printf("enc=[%s]\n", sEnc) sDec, err := base64.StdEncoding.DecodeString(sEnc) if err != nil { fmt.Printf("base64 decode failure, error=[%v]\n", err) } else { fmt.Printf("dec=[%s]\n", sDec) } return sEnc } //使用Stream load将文件数据导入到Doris对应的数据表中 func batch_load_file(load StreamLoad, file string) { client := &http.Client{} //生成要访问的url url := "http://10.220.146.10:8030/api/test_2/user_info/_stream_load" //fmt.Formatter(.Format(url,load.dbName,l)) fileContext, err := ioutil.ReadFile(file) if err != nil { log.Println("Failed to Read the File", file, err) } record := strings.NewReader(string(fileContext)) //提交请求 reqest, err := http.NewRequest(http.MethodPut, url, record) //增加header选项 reqest.Header.Add("Authorization", "basic "+auth(load)) reqest.Header.Add("EXPECT", "100-continue") var u1 = uuid.Must(uuid.NewV4()) reqest.Header.Add("label", u1.String()) reqest.Header.Add("column_separator", ",") if err != nil { panic(err) } //处理返回结果 response, _ := client.Do(reqest) if response.StatusCode == 200 { body, _ := ioutil.ReadAll(response.Body) responseBody := ResponseBody{} jsonStr := string(body) err := json.Unmarshal([]byte(jsonStr), &responseBody) if err != nil { fmt.Println(err.Error()) } if responseBody.Status == "Success" { //如果有被过滤的数据，打印错误的URL if responseBody.NumberFilteredRows > 0 { fmt.Printf("Error Data : %s ", responseBody.ErrorURL) } else { fmt.Printf("Success import data : %d", responseBody.NumberLoadedRows) } } fmt.Println(string(body)) } defer response.Body.Close() } //内存流数据，通过Stream Load导入Doris表中 func batch_load_data(load StreamLoad, data string) { client := &http.Client{} //生成要访问的url url := "http://10.220.146.10:8030/api/test_2/user_info/_stream_load" //fmt.Formatter(.Format(url,load.dbName,l)) record := strings.NewReader(data) //提交请求 reqest, err := http.NewRequest(http.MethodPut, url, record) //增加header选项 reqest.Header.Add("Authorization", "basic "+auth(load)) reqest.Header.Add("EXPECT", "100-continue") var u1 = uuid.Must(uuid.NewV4()) reqest.Header.Add("label", u1.String()) reqest.Header.Add("column_separator", ",") if err != nil { panic(err) } //处理返回结果 response, _ := client.Do(reqest) if response.StatusCode == 200 { body, _ := ioutil.ReadAll(response.Body) responseBody := ResponseBody{} jsonStr := string(body) err := json.Unmarshal([]byte(jsonStr), &responseBody) if err != nil { fmt.Println(err.Error()) } if responseBody.Status == "Success" { //如果有被过滤的数据，打印错误的URL if responseBody.NumberFilteredRows > 0 { fmt.Printf("Error Data : %s ", responseBody.ErrorURL) } else { fmt.Printf("Success import data : %d", responseBody.NumberLoadedRows) } } else { fmt.Printf("Error Message : %s \n", responseBody.Message) fmt.Printf("Error Data : %s ", responseBody.ErrorURL) } //fmt.Println(jsonStr) } defer response.Body.Close() } //获取BE列表 func get_doris_be_list() *list.List { var load StreamLoad load.userName = "root" load.password = "" client := &http.Client{} //生成要访问的url url := "http://10.220.146.10:8030/api/backends?is_alive=true" //提交请求 reqest, err := http.NewRequest("GET", url, nil) //增加header选项 reqest.Header.Add("Authorization", "basic "+auth(load)) if err != nil { panic(err) } //处理返回结果 response, _ := client.Do(reqest) bes := list.New() if response.StatusCode == 200 { body, _ := ioutil.ReadAll(response.Body) backends := Backend{} jsonStr := string(body) err := json.Unmarshal([]byte(jsonStr), &backends) if err != nil { fmt.Println(err.Error()) } for _, beinfo := range backends.Data.Backends { be := beinfo.IP + ":" + strconv.Itoa(beinfo.HTTPPort) bes.PushBack(be) } } defer response.Body.Close() return bes } //Stream load返回消息结构体 type ResponseBody struct { TxnID int `json:"TxnId"` Label string `json:"Label"` Status string `json:"Status"` Message string `json:"Message"` NumberTotalRows int `json:"NumberTotalRows"` NumberLoadedRows int `json:"NumberLoadedRows"` NumberFilteredRows int `json:"NumberFilteredRows"` NumberUnselectedRows int `json:"NumberUnselectedRows"` LoadBytes int `json:"LoadBytes"` LoadTimeMs int `json:"LoadTimeMs"` BeginTxnTimeMs int `json:"BeginTxnTimeMs"` StreamLoadPutTimeMs int `json:"StreamLoadPutTimeMs"` ReadDataTimeMs int `json:"ReadDataTimeMs"` WriteDataTimeMs int `json:"WriteDataTimeMs"` CommitAndPublishTimeMs int `json:"CommitAndPublishTimeMs"` ErrorURL string `json:"ErrorURL"` } //获取BE列表返回结构体 type Backend struct { Msg string `json:"msg"` Code int `json:"code"` Data struct { Backends []struct { IP string `json:"ip"` HTTPPort int `json:"http_port"` IsAlive bool `json:"is_alive"` } `json:"backends"` } `json:"data"` Count int `json:"count"` } func main() { var load StreamLoad load.userName = "root" load.password = "" //auth_info := auth(load) //fmt.Println(auth_info) //backends := get_doris_be_list() //for e := backends.Front(); e != nil; e = e.Next() { // fmt.Println(e.Value) //} data := "10001,张***,西安,30,1,133****760,陕西省**********,2021-03-12 12:34:12" batch_load_data(load, data) //batch_load_file(/load, "/Users/zhangfeng/Downloads/test.csv") }
