Flink stream load方式写入到doris遇到的问题

雨点打透心脏的1/2处 2022-11-07 05:53 723阅读 0赞

**1，Flink stream load 写入数据到doris的时候，偶然发现be节点挂掉的时候，我的yarn任务也会跟着报错，然后询问了开发小姐姐：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

继续跟小姐姐沟通：

![20210311154822184.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]

通过查看跟确认，发现tablet没有问题，这有尴尬了，只有继续测试，结果如下

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2][]

**2，这我就没办法理解了。我目前使用的是doris 0.13.15，遇到这个问题，是因为在通过http写入doris的时候，需要通过FE去查找BE节点的状态，还有个写入是直接写入到BE，但是如果BE节点挂了也会遇到类似的问题。**

**所以等待官网开发大佬的回复，也有可能是哪个参数跟姿势没解放，目前搁置吧先。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3][]

**3，这里最需要解决的是 ：Flink+http方式保证不了事务，doris自己可以保证原子性，如果一批数据写入失败，又要保证数据是有序的，BE节点挂掉，重启的这段时间数据怎么处理？**

当然，正常BE节点如果资源够的情况下，是不会挂的，官方小姐姐的原话（只能说生产还算稳定，数据量大，压力大也会挂）

4，目前flink写入doris 一定要开启checkpoint，要开启重试机制。然后如果写入异常，写入的算子部分要抛异常，导致任务失败，然后重试：

env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5, Time.of(10L, TimeUnit.SECONDS)));

5，有朋友使用stream load方式写入doris，遇到错误信息  
 java.net.SocketException: Connection reset by peer: socket write error

以为是写入受限制报错，其实不是的，是因为新版本0.15版本的原因，部分参数没添加（http参数）

代码如下：

import com.sjb.common.exceptions.BizException;
    import org.apache.commons.codec.binary.Base64;
    import org.apache.commons.lang3.StringUtils;
    import org.apache.http.HttpEntity;
    import org.apache.http.HttpHeaders;
    import org.apache.http.HttpVersion;
    import org.apache.http.client.config.RequestConfig;
    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.util.EntityUtils;
    
    import java.nio.charset.StandardCharsets;
    
    
    public class DorisStreamLoadDev {
        private final static String DORIS_HOST = "dev-hadoop";
        private final static String DORIS_USER = "root";
        private final static String DORIS_PASSWORD = "root";
        private final static int DORIS_HTTP_PORT = 8030;
    
    
        private static final String CHARSET = "UTF-8";
    
        private static final Integer HTTP_OK = 200;
    
        private static final String CONTENT_TYPE_NAME = "Content-Type";
    
        private static final String CONTENT_TYPE_JSON = "application/json;charset=UTF-8";
    
        private static final String CONTENT_TYPE_XML = "text/xml;charset=UTF-8";
    
        private static final String CONTENT_TYPE_FORM = "application/x-www-form-urlencoded;charset=UTF-8";
    
        private static final String ACCEPT_NAME = "Accept";
        private static final String ACCEPT = "application/json;charset=UTF-8";
    
        private static final int TIMEOUT = 60000;//这个要弄长点
    
        private RequestConfig defaultRequestConfig;
    
        private HttpClientBuilder httpClientBuilder;
        private CloseableHttpClient client;
    
        public DorisStreamLoadDev() {
            defaultRequestConfig = RequestConfig.custom()
                    .setSocketTimeout(TIMEOUT)
                    .setConnectTimeout(TIMEOUT)
                    .setConnectionRequestTimeout(TIMEOUT)
                    .build();
    
            this.httpClientBuilder = HttpClients
                    .custom()
                    .setRedirectStrategy(new DefaultRedirectStrategy() {
                        @Override
                        protected boolean isRedirectable(String method) {
                            return true;
                        }
                    });
    
            this.client = httpClientBuilder.build();
    
        }
    
    
        public void sendData(String content, String db, String table) {
    
            String loadResult = "";
            CloseableHttpResponse response = null;
            final String loadUrl = String.format("http://%s:%s/api/%s/%s/_stream_load",
                    DORIS_HOST,
                    DORIS_HTTP_PORT,
                    db,
                    table);
            try {
    
                HttpPut httpPut = new HttpPut(loadUrl);
                StringEntity entity = new StringEntity(content, "UTF-8");
                httpPut.setHeader(HttpHeaders.EXPECT, "100-continue");
                httpPut.setHeader(HttpHeaders.AUTHORIZATION, basicAuthHeader(DORIS_USER, DORIS_PASSWORD));
                httpPut.setHeader("strip_outer_array", "true");
                httpPut.setHeader("format", "json");
                httpPut.setHeader("merge_type", "MERGE");
                httpPut.setHeader("delete", "canal_type=\"DELETE\"");
                httpPut.setProtocolVersion(HttpVersion.HTTP_1_0);
                httpPut.addHeader(CONTENT_TYPE_NAME, CONTENT_TYPE_FORM);
                httpPut.addHeader(ACCEPT_NAME, ACCEPT);
    //            httpPut.addHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_CLOSE);
    
    
                //todo 输出参数
                httpPut.setConfig(defaultRequestConfig);
                httpPut.setEntity(entity);
                response = client.execute(httpPut);
                HttpEntity httpEntity = response.getEntity();
    
                //todo 调用方法
                if (httpEntity != null) {
                    loadResult = EntityUtils.toString(httpEntity);
                }
                final int statusCode = response.getStatusLine().getStatusCode();
    
                EntityUtils.consume(httpEntity);
    
                if (statusCode != 200) {
    //                System.out.println("loadResult = " + loadResult);
                    throw new BizException("写入失败，loadResult=" + loadResult);
                } else {
                    if (loadResult.contains("OK") && loadResult.contains("Success")) {
                        //todo 正式环境要去掉
                        System.out.println("写入成功，loadResult = " + loadResult);
                    } else {
    //                    System.out.println("写入失败，loadResult=" + loadResult);
                        throw new BizException("写入失败，loadResult=" + loadResult);
    
                    }
                }
    
            } catch (Exception ex) {
    
                // 数组转字符串(逗号分隔)(推荐)
                String error_info = StringUtils.join(ex.getStackTrace(), "\n");
                throw new BizException("[DorisStreamLoadProd]出错，异常信息：[" + ex.toString() + "],具体错误地址: [" + error_info + "]");
    
            } finally {
                if (null != response) {
                    try {
                        response.close();
                    } catch (Exception ex) {
                        String error_info = StringUtils.join(ex.getStackTrace(), "\n");
                        throw new BizException("[DorisStreamLoadProd],response关闭异常,异常信息：[" + ex.toString() + "],具体错误地址: [" + error_info + "]");
                    }
                }
            }
        }
    
    
        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);
        }
    
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20221023/2e6189e596924da19ac6f045d9af0a34.png
[20210311154822184.png]: /images/20221023/6fd193ab0b44474dbb68b774a1ae5ce5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: /images/20221023/4e2ed8636df041c7b32698f96cc931db.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2]: /images/20221023/73ff8919dc2545ce8d7e9296b069ffdf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3]: /images/20221023/29f645bce7f349ce8bcb177b7a06dbb8.png