Doris数据库--导入数据之 stream load Java调FE接口上传文件导入数据 ...

需求配景

2023年11月份接到需求：将8张表数据文件导入到Doris中,数据文件是txt格式存放在zip内,平均天天数据量80亿,占磁盘空间300G
以前从来没接触过Doris数据库,所以初版用的jdbc批量导入,也用了一些代码层面的导入优化,效果不抱负
查看Doris官网后发现有很多种导入方式

经过实战streamLoad方式导入服从快又稳，已经稳定运行三个月
效果图

正题---代码部分

工具类

1. package io.geekidea.boot.utils;
3. import cn.hutool.core.io.FileTypeUtil;
4. import cn.hutool.core.io.FileUtil;
5. import cn.hutool.json.JSONObject;
6. import lombok.extern.slf4j.Slf4j;
7. import org.apache.commons.codec.binary.Base64;
8. import org.apache.http.HttpHeaders;
9. import org.apache.http.client.methods.CloseableHttpResponse;
10. import org.apache.http.client.methods.HttpPut;
11. import org.apache.http.entity.FileEntity;
12. import org.apache.http.impl.client.CloseableHttpClient;
13. import org.apache.http.impl.client.DefaultRedirectStrategy;
14. import org.apache.http.impl.client.HttpClientBuilder;
15. import org.apache.http.impl.client.HttpClients;
16. import org.apache.http.util.EntityUtils;
18. import java.io.File;
19. import java.io.IOException;
20. import java.nio.charset.StandardCharsets;
22. /*
23. 这是一个 Doris Stream Load 示例，需要依赖
24. <dependency>
25.     <groupId>org.apache.httpcomponents</groupId>
26.     <artifactId>httpclient</artifactId>
27.     <version>4.5.13</version>
28. </dependency>
29. */
30. @Slf4j
31. public class DorisStreamLoader implements Runnable {
33.     //可以选择填写 FE 地址以及 FE 的 http_port，但须保证客户端和 BE 节点的连通性。
34.     private String HOST;
35.     private int PORT;
36.     // 要导入的数据库
37.     private String DATABASE;
38.     // Doris 密码
39.     private String username;
40.     // Doris 用户名
41.     private String password;
42.     private static String TABLE = "";     // 要导入的表
43.     private static String COLUMNS = "";     // 要导入的表字段顺序
44.     private static File sourceFile;//源文件
45.     private static String toFilePath;//目标文件路径
46.     private static String errorPath;//失败文件路径
47.     private static String Separator;//目标文件路径
48.     private static String LOAD_FILE_NAME = "/path/to/1.txt"; // 要导入的本地文件路径
50.     public DorisStreamLoader(String host,int port,String database,String username,String password,String table, String columns,File sourceFile,String toFilePath,String Separator,String errorPath) {
51.         this.HOST = host;
52.         this.PORT = port;
53.         this.DATABASE = database;
54.         this.username = username;
55.         this.password = password;
56.         this.TABLE = table;
57.         this.COLUMNS = columns;
58.         this.sourceFile = sourceFile;
59.         this.toFilePath = toFilePath;
60.         this.Separator = Separator;
61.         this.errorPath = errorPath;
62.     }
64.     private final static HttpClientBuilder httpClientBuilder = HttpClients
65.             .custom()
66.             .setRedirectStrategy(new DefaultRedirectStrategy() {
67.                 @Override
68.                 protected boolean isRedirectable(String method) {
69.                     // 如果连接目标是 FE，则需要处理 307 redirect。
70.                     return true;
71.                 }
72.             });
76.     @Override
77.     public void run() {
78.         load(sourceFile,toFilePath);
79.     }
81.     /**
82.      * doris stream load
83.      * @param file 导入文件
84.      * @param toFile 导入后移动路径
85.      */
86.     public void load(File file,String toFile) {
87.         try (CloseableHttpClient client = httpClientBuilder.build()) {
88.             HttpPut put = new HttpPut(String.format("http://%s:%s/api/%s/%s/_stream_load",
89.                     HOST, PORT, DATABASE, TABLE));
90.             put.setHeader(HttpHeaders.EXPECT, "100-continue");
91.             put.setHeader(HttpHeaders.AUTHORIZATION, basicAuthHeader(username, password));
92.             // 可以在 Header 中设置 stream load 相关属性，这里我们设置 label 和 column_separator。
93.             // 不设置时候 Doris会生成默认的
94.             //put.setHeader("label","label1");
95.             String type = FileTypeUtil.getType(file);
96.             put.setHeader("format",type);
97.             put.setHeader("columns",COLUMNS);
98.             // 设置错误行过滤比例
99.             put.setHeader("max_filter_ratio","0.1");
100.             put.setHeader("column_separator",Separator);
101.             // 设置导入文件。
102.             // 这里也可以使用 StringEntity 来传输任意数据。
103.             FileEntity entity = new FileEntity(file);
104.             put.setEntity(entity);
105.             log.info("Put request: {}==={}==={}",file.getName(), String.format("http://%s:%s/api/%s/%s/_stream_load",
106.                     HOST, PORT, DATABASE, TABLE),COLUMNS);
107.             try (CloseableHttpResponse response = client.execute(put)) {
108.                 String loadResult = "";
109.                 if (response.getEntity() != null) {
110.                     loadResult = EntityUtils.toString(response.getEntity());
111.                 }
113.                 final int statusCode = response.getStatusLine().getStatusCode();
114.                 if (statusCode != 200) {
115.                     throw new IOException(
116.                             String.format("导入失败. status: %s load result: %s", statusCode, loadResult));
117.                 }else {
118.                     JSONObject json = new JSONObject(loadResult);
119.                     if(json.getStr("Status").equals("Success") || json.getStr("Status").contains("Publish")){
120.                         FileUtil.move(file,new File(toFile),true);
121.                         log.info("导入成功");
122.                     }else {
123.                         FileUtil.move(file,new File(toFile),true);
124.                         log.info("导入失败");
125.                     }
126.                 }
127.                 log.info("Get load result: {}", loadResult);
128.             }
129.         }catch (Exception e) {
130.             FileUtil.move(file,new File(toFile),true);
131.             log.error("上传文件：{}到Doris报错{}",file.getName(),e);
132.         }
133.     }
135.     private String basicAuthHeader(String username, String password) {
136.         final String tobeEncode = username + ":" + password;
137.         //log.info("用户名：密码{}",tobeEncode);
138.         byte[] encoded = Base64.encodeBase64(tobeEncode.getBytes(StandardCharsets.UTF_8));
139.         return "Basic " + new String(encoded);
140.     }
141.     public static void main(String[] args) throws Exception{
142.         //DorisStreamLoader loader = new DorisStreamLoader();
143.         //File file = new File(LOAD_FILE_NAME);
144.         //loader.load(file,"/app/file");
146.     }
147. }

复制代码

具体使用

此处可以自行更改为并发提交：假如提交过猛会导入失败，根据自己数据库集群性能更改。

1. public void Import(String folders) {
2.         //文件所在目录：folders + unzipTargetPath
3.         //cn.hutool.core.io.FileUtil;
4.         List<File> unzipTxt = FileUtil.loopFiles(folders + unzipTargetPath);
5.         log.info("需要导入文件数量{}", unzipTxt.size());
6.         for (File file : unzipTxt) {
7.             Map<String, String> map = getColumn(file.getName());
8.             new DorisStreamLoader(
9.                     sdkConfig.getTargetIp(),
10.                     sdkConfig.getTargetPort(),
11.                     sdkConfig.getTargetDatabase(),
12.                     sdkConfig.getTargetUsername(),
13.                     sdkConfig.getTargetPassword(),
14.                     map.get("table"), map.get("columns"), file, folders + movePath + file.getName(),
15.                     "\\u0002",
16.                     folders + errorPath + file.getName()).run();
17.         }
18.     }
20.     //获取表名和字段
21.     private Map<String, String> getColumn(String path) {
22.         Map<String, String> map = new HashMap<>();
23.         if (path.contains("ip")) {
24.             map.put("table", "tableName");
25.             map.put("columns", "A,B,C,D,E");
26.         }
27.         //省略。。。。
28.         return map;
29.     }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。