|
|
@@ -1,94 +0,0 @@
|
|
|
-package cn.com.ctop.job.adq.utils.doris.sink;
|
|
|
-
|
|
|
-import cn.com.ctop.job.adq.data.utils.JsonUtils;
|
|
|
-import com.alibaba.fastjson.JSONObject;
|
|
|
-import lombok.extern.slf4j.Slf4j;
|
|
|
-import org.apache.http.HttpEntity;
|
|
|
-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.HttpClients;
|
|
|
-import org.apache.http.util.EntityUtils;
|
|
|
-
|
|
|
-import java.nio.charset.StandardCharsets;
|
|
|
-import java.util.Base64;
|
|
|
-import java.util.List;
|
|
|
-import java.util.UUID;
|
|
|
-import java.util.stream.Collectors;
|
|
|
-
|
|
|
-@Slf4j
|
|
|
-public class SimpleHttpDorisSinkClient implements DorisSinkClient {
|
|
|
- private final String feHost; // http://fe_host:8030
|
|
|
- private final String user; // doris 用户名
|
|
|
- private final String password; // doris 密码
|
|
|
- private final String format; // json/csv, 推荐 json
|
|
|
- private final String labelPrefix; // stream load label 前缀(可选)
|
|
|
-
|
|
|
- public SimpleHttpDorisSinkClient(String feHost, String user, String password, String labelPrefix) {
|
|
|
- this.feHost = feHost;
|
|
|
- this.user = user;
|
|
|
- this.password = password;
|
|
|
- this.format = "json";
|
|
|
- this.labelPrefix = labelPrefix;
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public void streamLoad(List<RecordWithMeta> records) throws Exception {
|
|
|
- if (records == null || records.isEmpty()) return;
|
|
|
-
|
|
|
- RecordWithMeta sample = records.get(0);
|
|
|
- String db = sample.getDatabase();
|
|
|
- String table = sample.getTable();
|
|
|
-
|
|
|
- String label = labelPrefix + "_" + db + "_" + table + "_" + UUID.randomUUID();
|
|
|
- String url = String.format("%s/api/%s/%s/_stream_load", feHost, db, table);
|
|
|
-
|
|
|
- // 提取 data
|
|
|
- List<String> dataList = records.stream()
|
|
|
- .map(d->{
|
|
|
- Object data = d.getData();
|
|
|
- if (data instanceof JSONObject) {
|
|
|
- return data.toString();
|
|
|
- }else {
|
|
|
- return JsonUtils.toFilterJSON(data);
|
|
|
- }
|
|
|
- })
|
|
|
- .collect(Collectors.toList());
|
|
|
-
|
|
|
- String jsonBody = dataList.toString();
|
|
|
- //log.info("jsonBody:{}", jsonBody);
|
|
|
-
|
|
|
- HttpPut put = new HttpPut(url);
|
|
|
- put.setHeader("Expect", "100-continue");
|
|
|
- put.setHeader("label", label);
|
|
|
- put.setHeader("Content-Type", "application/json");
|
|
|
- if (user != null && password != null) {
|
|
|
- String auth = Base64.getEncoder().encodeToString((user + ":" + password).getBytes(StandardCharsets.UTF_8));
|
|
|
- put.setHeader("Authorization", "Basic " + auth);
|
|
|
- }
|
|
|
-
|
|
|
- put.setHeader("format", format); // json
|
|
|
- put.setHeader("strip_outer_array", "true"); // json array
|
|
|
- put.setHeader("fuzzy_parse", "true"); // json array
|
|
|
-
|
|
|
- put.setEntity(new StringEntity(jsonBody, StandardCharsets.UTF_8));
|
|
|
-
|
|
|
- try (CloseableHttpClient httpClient = HttpClients.custom()
|
|
|
- .setRedirectStrategy(new StrictRedirectStrategy()) // 支持 307/301 重定向
|
|
|
- .build();
|
|
|
- CloseableHttpResponse response = httpClient.execute(put)) {
|
|
|
- int code = response.getStatusLine().getStatusCode();
|
|
|
- HttpEntity entity = response.getEntity();
|
|
|
- String responseBody = EntityUtils.toString(entity, "UTF-8");
|
|
|
- if (code != 200 && code != 201) {
|
|
|
- throw new RuntimeException("Doris stream load failed. Status: " + code + ", response: " + responseBody);
|
|
|
- }
|
|
|
- // 可选: 解析返回内容判断 success
|
|
|
- if (!responseBody.contains("\"Status\": \"Success\"")) {
|
|
|
- throw new RuntimeException("Stream load returned non-success: " + responseBody);
|
|
|
- }
|
|
|
- log.info("Doris stream load success: {}", responseBody);
|
|
|
- }
|
|
|
- }
|
|
|
-}
|