Stream load doris

代码如下：经测试有效

package Demo;


import com.alibaba.fastjson.JSONObject;
import org.apache.commons.net.util.Base64;
import org.apache.http.HttpHeaders;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.DefaultRedirectStrategy;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.protocol.RequestContent;
import org.apache.http.util.EntityUtils;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;

public class DorisStreamLoader {

private  final String user;
private final String password;

private final String loadUrl;

public static void main(String[] args) throws Exception {
    /*ArrayList<JSONObject> list = new ArrayList<>();
    JSONObject jsonObject = new JSONObject();
    jsonObject.put("waybillNumber","a123456");
    list.add(jsonObject );
	JSONObject jsonObject2 = new JSONObject();
    jsonObject2.put("waybillNumber","789");
    list.add(jsonObject2 );
    String jsonData = list.toString();
    System.out.println(jsonData);
    String s = streamLoader.loadJson(jsonData);
*/

    DorisStreamLoader streamLoader = new DorisStreamLoader(host, port, "database", "table", "user", "password");
    JSONObject jsonObject = new JSONObject();
    jsonObject.put("waybillNumber","a123456");
    String jsonData = jsonObject.toString();
    String s = streamLoader.loadJson(jsonData);
    JSONObject jsonObject1 = JSONObject.parseObject(s);
    String status = jsonObject1.getString("Status");

    if(status.equalsIgnoreCase("Success") || status.equalsIgnoreCase("Publish Timeout")){
        System.out.println("load Success");
    }else {
        s = streamLoader.loadJson(jsonData);
        if(! status.equalsIgnoreCase("Success") || status.equalsIgnoreCase("Publish Timeout") ){
            System.out.println("load Fail"+jsonData.toString());
        }
    }
}

public DorisStreamLoader(String host, int port, String database, String table, String user, String password) {
    this.user = user;
    this.password = password;
    this.loadUrl = String.format("http://%s:%s/api/%s/%s/_stream_load", host, port, database, table);
}


private final static HttpClientBuilder httpClientBuilder = HttpClients
        .custom()
        .setRedirectStrategy(new DefaultRedirectStrategy() {
            @Override
            protected boolean isRedirectable(String method) {
                // If the connection target is FE, you need to deal with 307 redirect。
                return true;
            }
        }).addInterceptorLast(new RequestContent(true));


public String loadJson(String jsonData) throws Exception {
    String loadResult = "";
    try (CloseableHttpClient client = httpClientBuilder.build()) {
        HttpPut put = new HttpPut(loadUrl);
       /* put.removeHeaders(HttpHeaders.CONTENT_LENGTH);
        put.removeHeaders(HttpHeaders.TRANSFER_ENCODING);*/

        put.setHeader(HttpHeaders.EXPECT, "100-continue");
        put.setHeader(HttpHeaders.AUTHORIZATION, basicAuthHeader(user, password));

        put.setHeader("label", UUID.randomUUID().toString());
        put.setHeader("column_separator", ",");
        put.setHeader("format", "json");
         // strip_outer_array: 布尔类型，为true表示json数据以数组对象开始且将数组对象中进行展平，默认值是false
         //这个功能要求 Array 中的每行数据的字段顺序完全一致。Doris 仅会根据第一行的字段顺序做解析，然后以下标的形式访问之后的数据。
        put.setHeader("strip_outer_array", "true");


        StringEntity entity = new StringEntity(jsonData,"UTF-8");
        put.setEntity(entity);

        try (CloseableHttpResponse response = client.execute(put)) {

            if (response.getEntity() != null) {
                loadResult = EntityUtils.toString(response.getEntity());
            }

            final int statusCode = response.getStatusLine().getStatusCode();
            if (statusCode != 200) {
                throw new IOException(String.format("Stream load failed. status: %s load result: %s", statusCode, loadResult));
            }

//                System.out.println("Get load result: " + loadResult);
        }
    }
    return loadResult;
}

/**
 * 封装认证信息
 *
 * @param username String
 * @param password String
 * @return String
 */
private String basicAuthHeader(String username, String password) {
    final String tobeEncode = username + ":" + password;
    byte[] encoded = Base64.encodeBase64(tobeEncode.getBytes(StandardCharsets.UTF_8));
    return "Basic " + new String(encoded);
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127