• DataX DorisWriter 插件DorisStreamLoadObserver类详细解读

    DorisStreamLoadObserver 类是一个用于将数据加载到 Doris（以前称为 Palo）数据库中并监视加载过程的 Java 类。该类提供了一组方法，用于构建 HTTP 请求、处理 HTTP 响应以及监控数据加载的状态。以下是每个方法的具体作用：

    1. DorisStreamLoadObserver(Keys options): 这是类的构造函数，用于初始化加载数据所需的配置选项。
    2. void streamLoad(WriterTuple data) throws Exception: 该方法是数据加载的主要方法。它将给定的数据（WriterTuple 对象）加载到 Doris 数据库中。它构建了用于将数据发送到 Doris 的 HTTP 请求，并根据响应状态来确定加载是否成功。如果加载失败，它会抛出异常。
    3. private void checkStreamLoadState(String host, String label) throws IOException: 这个方法用于检查数据加载的状态。它会不断地轮询 Doris 服务器，以获取特定加载任务的最终状态。根据加载状态的不同，它可能会抛出异常或者在加载完成时返回。
    4. private byte[] addRows(List rows, int totalBytes): 此方法根据给定的数据行和总字节数，构建用于加载的字节数组。它根据配置中的数据格式（CSV 或 JSON）将数据行连接起来，并添加适当的分隔符。
    5. private Map put(String loadUrl, String label, byte[] data) throws IOException: 该方法执行 HTTP PUT 请求，将数据加载到 Doris 数据库中。它构建了包含数据的请求实体，发送到指定的加载 URL，并解析响应以获取加载结果。
    6. private String getBasicAuthHeader(String username, String password): 此方法用于生成基本身份验证头部，以便在 HTTP 请求中进行身份验证。
    7. private HttpEntity getHttpEntity(CloseableHttpResponse response): 这是一个实用方法，用于从 HTTP 响应中提取实体内容。
    8. private String getLoadHost(): 该方法从配置选项中获取用于加载数据的主机地址列表，并尝试连接到这些主机以检查其可用性。它会返回第一个可用的主机地址。

    DorisStreamLoadObserver 类主要用于处理数据加载任务，它负责构建适当的 HTTP 请求，将数据发送到 Doris 数据库，并监控加载任务的状态。通过这些方法，可以实现将数据从外部系统加载到 Doris 数据库中，并在加载过程中进行必要的状态检查和错误处理。

    1. import org.apache.commons.codec.binary.Base64;
    2. import org.apache.http.HttpEntity;
    3. import org.apache.http.HttpHeaders;
    4. import org.apache.http.client.config.RequestConfig;
    5. import org.apache.http.client.methods.CloseableHttpResponse;
    6. import org.apache.http.client.methods.HttpGet;
    7. import org.apache.http.client.methods.HttpPut;
    8. import org.apache.http.entity.ByteArrayEntity;
    9. import org.apache.http.impl.client.CloseableHttpClient;
    10. import org.apache.http.impl.client.DefaultRedirectStrategy;
    11. import org.apache.http.impl.client.HttpClientBuilder;
    12. import org.apache.http.impl.client.HttpClients;
    13. import org.apache.http.util.EntityUtils;
    14. import org.json.simple.JSONValue;
    15. import org.slf4j.Logger;
    16. import org.slf4j.LoggerFactory;
    17. import java.io.IOException;
    18. import java.net.HttpURLConnection;
    19. import java.net.URL;
    20. import java.nio.ByteBuffer;
    21. import java.nio.charset.StandardCharsets;
    22. import java.util.HashMap;
    23. import java.util.List;
    24. import java.util.Map;
    25. import java.util.concurrent.TimeUnit;
    26. import java.util.stream.Collectors;
    27.  
    28. public class DorisStreamLoadObserver {
    29.     private static final Logger LOG = LoggerFactory.getLogger(DorisStreamLoadObserver.class);
    30.  
    31.     private Keys options;
    32.  
    33.     private long pos;
    34.     private static final String RESULT_FAILED = "Fail";
    35.     private static final String RESULT_LABEL_EXISTED = "Label Already Exists";
    36.     private static final String LAEBL_STATE_VISIBLE = "VISIBLE";
    37.     private static final String LAEBL_STATE_COMMITTED = "COMMITTED";
    38.     private static final String RESULT_LABEL_PREPARE = "PREPARE";
    39.     private static final String RESULT_LABEL_ABORTED = "ABORTED";
    40.     private static final String RESULT_LABEL_UNKNOWN = "UNKNOWN";
    41.  
    42.     public DorisStreamLoadObserver(Keys options) {
    43.         this.options = options;
    44.     }
    45.  
    46.     // 数据写入 Doris 的主要方法
    47.     public void streamLoad(WriterTuple data) throws Exception {
    48.         String host = getLoadHost();
    49.         if (host == null) {
    50.             throw new IOException("load_url cannot be empty, or the host cannot connect. Please check your configuration.");
    51.         }
    52.         String loadUrl = new StringBuilder(host)
    53.                 .append("/api/")
    54.                 .append(options.getDatabase())
    55.                 .append("/")
    56.                 .append(options.getTable())
    57.                 .append("/_stream_load")
    58.                 .toString();
    59.         LOG.info("Start to join batch data: rows[{}] bytes[{}] label[{}].", data.getRows().size(), data.getBytes(), data.getLabel());
    60.         Map<String, Object> loadResult = put(loadUrl, data.getLabel(), addRows(data.getRows(), data.getBytes().intValue()));
    61.         LOG.info("StreamLoad response :{}", JSONValue.toJSONString(loadResult));
    62.         final String keyStatus = "Status";
    63.         if (null == loadResult || !loadResult.containsKey(keyStatus)) {
    64.             throw new IOException("Unable to flush data to Doris: unknown result status.");
    65.         }
    66.         LOG.debug("StreamLoad response:{}", JSONValue.toJSONString(loadResult));
    67.         if (RESULT_FAILED.equals(loadResult.get(keyStatus))) {
    68.             throw new IOException(
    69.                     new StringBuilder("Failed to flush data to Doris.\n").append(JSONValue.toJSONString(loadResult)).toString()
    70.             );
    71.         } else if (RESULT_LABEL_EXISTED.equals(loadResult.get(keyStatus))) {
    72.             LOG.debug("StreamLoad response:{}", JSONValue.toJSONString(loadResult));
    73.             checkStreamLoadState(host, data.getLabel());
    74.         }
    75.     }
    76.  
    77.     // 检查数据加载状态的方法
    78.     private void checkStreamLoadState(String host, String label) throws IOException {
    79.         int idx = 0;
    80.         while (true) {
    81.             try {
    82.                 TimeUnit.SECONDS.sleep(Math.min(++idx, 5));
    83.             } catch (InterruptedException ex) {
    84.                 break;
    85.             }
    86.             try (CloseableHttpClient httpclient = HttpClients.createDefault()) {
    87.                 HttpGet httpGet = new HttpGet(new StringBuilder(host).append("/api/").append(options.getDatabase()).append("/get_load_state?label=").append(label).toString());
    88.                 httpGet.setHeader("Authorization", getBasicAuthHeader(options.getUsername(), options.getPassword()));
    89.                 httpGet.setHeader("Connection", "close");
    90.  
    91.                 try (CloseableHttpResponse resp = httpclient.execute(httpGet)) {
    92.                     HttpEntity respEntity = getHttpEntity(resp);
    93.                     if (respEntity == null) {
    94.                         throw new IOException(String.format("Failed to flush data to Doris, Error " +
    95.                                 "could not get the final state of label[%s].\n", label), null);
    96.                     }
    97.                     Map<String, Object> result = (Map<String, Object>) JSONValue.parse(EntityUtils.toString(respEntity));
    98.                     String labelState = (String) result.get("state");
    99.                     if (null == labelState) {
    100.                         throw new IOException(String.format("Failed to flush data to Doris, Error " +
    101.                                 "could not get the final state of label[%s]. response[%s]\n", label, EntityUtils.toString(respEntity)), null);
    102.                     }
    103.                     LOG.info(String.format("Checking label[%s] state[%s]\n", label, labelState));
    104.                     switch (labelState) {
    105.                         case LAEBL_STATE_VISIBLE:
    106.                         case LAEBL_STATE_COMMITTED:
    107.                             return;
    108.                         case RESULT_LABEL_PREPARE:
    109.                             continue;
    110.                         case RESULT_LABEL_ABORTED:
    111.                             throw new DorisWriterExcetion(String.format("Failed to flush data to Doris, Error " +
    112.                                     "label[%s] state[%s]\n", label, labelState), null, true);
    113.                         case RESULT_LABEL_UNKNOWN:
    114.                         default:
    115.                             throw new IOException(String.format("Failed to flush data to Doris, Error " +
    116.                                     "label[%s] state[%s]\n", label, labelState), null);
    117.                     }
    118.                 }
    119.             }
    120.         }
    121.     }
    122.  
    123.     // 根据格式将数据行拼接成字节数组
    124.     private byte[] addRows(List<byte[]> rows, int totalBytes) {
    125.         if (Keys.StreamLoadFormat.CSV.equals(options.getStreamLoadFormat())) {
    126.             Map<String, Object> props = (options.getLoadProps() == null ? new HashMap<>() : options.getLoadProps());
    127.             byte[] lineDelimiter = DelimiterParser.parse((String) props.get("line_delimiter"), "\n").getBytes(StandardCharsets.UTF_8);
    128.             ByteBuffer bos = ByteBuffer.allocate(totalBytes + rows.size() * lineDelimiter.length);
    129.             for (byte[] row : rows) {
    130.                 bos.put(row);
    131.                 bos.put(lineDelimiter);
    132.             }
    133.             return bos.array();
    134.         }
    135.  
    136.         if (Keys.StreamLoadFormat.JSON.equals(options.getStreamLoadFormat())) {
    137.             ByteBuffer bos = ByteBuffer.allocate(totalBytes + (rows.isEmpty() ? 2 : rows.size() + 1));
    138.             bos.put("[".getBytes(StandardCharsets.UTF_8));
    139.             byte[] jsonDelimiter = ",".getBytes(StandardCharsets.UTF_8);
    140.             boolean isFirstElement = true;
    141.             for (byte[] row : rows) {
    142.                 if (!isFirstElement) {
    143.                     bos.put(jsonDelimiter);
    144.                 }
    145.                 bos.put(row);
    146.                 isFirstElement = false;
    147.             }
    148.             bos.put("]".getBytes(StandardCharsets.UTF_8));
    149.             return bos.array();
    150.         }
    151.         throw new RuntimeException("Failed to join rows data, unsupported `format` from stream load properties:");
    152.     }
    153.  
    154. private Map<String, Object> put(String loadUrl, String label, byte[] data) throws IOException {
    155.         RequestConfig requestConfig = RequestConfig.custom()
    156.                 .setSocketTimeout(120 * 1000)
    157.                 .setConnectTimeout(120 * 1000)
    158.                 .setConnectionRequestTimeout(120 * 1000)
    159.                 .build();
    160.         try (CloseableHttpClient httpclient = HttpClientBuilder.create()
    161.                 .setDefaultRequestConfig(requestConfig)
    162.                 .setRedirectStrategy(new DefaultRedirectStrategy())
    163.                 .build()) {
    164.             HttpPut httpPut = new HttpPut(loadUrl);
    165.             httpPut.setHeader(HttpHeaders.CONTENT_TYPE, "application/octet-stream");
    166.             httpPut.setHeader("Authorization", getBasicAuthHeader(options.getUsername(), options.getPassword()));
    167.             httpPut.setEntity(new ByteArrayEntity(data));
    168.             try (CloseableHttpResponse resp = httpclient.execute(httpPut)) {
    169.                 HttpEntity respEntity = getHttpEntity(resp);
    170.                 if (respEntity == null) {
    171.                     throw new IOException("Failed to flush data to Doris, Error could not get the response entity.");
    172.                 }
    173.                 return (Map<String, Object>) JSONValue.parse(EntityUtils.toString(respEntity));
    174.             }
    175.         }
    176.     }
    177.  
    178.     // 构造 HTTP 请求中的基本认证头部
    179.     private String getBasicAuthHeader(String username, String password) {
    180.         String credentials = username + ":" + password;
    181.         byte[] credentialsBytes = credentials.getBytes(StandardCharsets.UTF_8);
    182.         String base64Credentials = Base64.encodeBase64String(credentialsBytes);
    183.         return "Basic " + base64Credentials;
    184.     }
    185.  
    186.     // 从 HTTP 响应中获取实体内容
    187.     private HttpEntity getHttpEntity(CloseableHttpResponse response) {
    188.         if (response != null) {
    189.             return response.getEntity();
    190.         }
    191.         return null;
    192.     }
    193.  
    194.     // 获取用于加载数据的主机地址
    195.     private String getLoadHost() {
    196.         List<String> hosts = options.getDorisStreamLoadUrls();
    197.         for (String host : hosts) {
    198.             try {
    199.                 HttpURLConnection connection = (HttpURLConnection) new URL(host).openConnection();
    200.                 connection.setRequestMethod("HEAD");
    201.                 int responseCode = connection.getResponseCode();
    202.                 if (responseCode == HttpURLConnection.HTTP_OK) {
    203.                     return host;
    204.                 }
    205.             } catch (IOException e) {
    206.                 LOG.warn("Failed to connect to host: {}", host);
    207.             }
    208.         }
    209.         return null;
    210.     }
    211. }
  • 相关阅读:
    基于nodejs+vue百鸟全科赏析网站
    911. 在线选举
    NFT 智能合约实战-快速开始（1）NFT发展历史 | NFT合约标准(ERC-721、ERC-1155和ERC-998)介绍
    C++面试八股文：C++中，函数的参数应该传值还是传引用？
    腾讯云Redis全面升级，性能提升400%，可用性高达5个9
    Android逆向学习（番外一）smali2java部分文件无法反编译的bug与修复方法
    Mysql事务
    软考高级软件架构师学习笔记一
    HDFS、YARN、MapReduce概述及三者之间的关系
    关于static修饰的成员方法成员成员变量的相关讨论
  • 原文地址：https://blog.csdn.net/linweidong/article/details/132634622