doris – 第2页 – gitweixin

doris, Java 8月 29,2023

DataX DorisWriter 插件DorisStreamLoadObserver类详细解读

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

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

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

import org.apache.commons.codec.binary.Base64;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHeaders;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.ByteArrayEntity;
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 org.json.simple.JSONValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public class DorisStreamLoadObserver {
    private static final Logger LOG = LoggerFactory.getLogger(DorisStreamLoadObserver.class);

    private Keys options;

    private long pos;
    private static final String RESULT_FAILED = "Fail";
    private static final String RESULT_LABEL_EXISTED = "Label Already Exists";
    private static final String LAEBL_STATE_VISIBLE = "VISIBLE";
    private static final String LAEBL_STATE_COMMITTED = "COMMITTED";
    private static final String RESULT_LABEL_PREPARE = "PREPARE";
    private static final String RESULT_LABEL_ABORTED = "ABORTED";
    private static final String RESULT_LABEL_UNKNOWN = "UNKNOWN";

    public DorisStreamLoadObserver(Keys options) {
        this.options = options;
    }

    // 数据写入 Doris 的主要方法
    public void streamLoad(WriterTuple data) throws Exception {
        String host = getLoadHost();
        if (host == null) {
            throw new IOException("load_url cannot be empty, or the host cannot connect. Please check your configuration.");
        }
        String loadUrl = new StringBuilder(host)
                .append("/api/")
                .append(options.getDatabase())
                .append("/")
                .append(options.getTable())
                .append("/_stream_load")
                .toString();
        LOG.info("Start to join batch data: rows[{}] bytes[{}] label[{}].", data.getRows().size(), data.getBytes(), data.getLabel());
        Map<String, Object> loadResult = put(loadUrl, data.getLabel(), addRows(data.getRows(), data.getBytes().intValue()));
        LOG.info("StreamLoad response :{}", JSONValue.toJSONString(loadResult));
        final String keyStatus = "Status";
        if (null == loadResult || !loadResult.containsKey(keyStatus)) {
            throw new IOException("Unable to flush data to Doris: unknown result status.");
        }
        LOG.debug("StreamLoad response:{}", JSONValue.toJSONString(loadResult));
        if (RESULT_FAILED.equals(loadResult.get(keyStatus))) {
            throw new IOException(
                    new StringBuilder("Failed to flush data to Doris.\n").append(JSONValue.toJSONString(loadResult)).toString()
            );
        } else if (RESULT_LABEL_EXISTED.equals(loadResult.get(keyStatus))) {
            LOG.debug("StreamLoad response:{}", JSONValue.toJSONString(loadResult));
            checkStreamLoadState(host, data.getLabel());
        }
    }

    // 检查数据加载状态的方法
    private void checkStreamLoadState(String host, String label) throws IOException {
        int idx = 0;
        while (true) {
            try {
                TimeUnit.SECONDS.sleep(Math.min(++idx, 5));
            } catch (InterruptedException ex) {
                break;
            }
            try (CloseableHttpClient httpclient = HttpClients.createDefault()) {
                HttpGet httpGet = new HttpGet(new StringBuilder(host).append("/api/").append(options.getDatabase()).append("/get_load_state?label=").append(label).toString());
                httpGet.setHeader("Authorization", getBasicAuthHeader(options.getUsername(), options.getPassword()));
                httpGet.setHeader("Connection", "close");

                try (CloseableHttpResponse resp = httpclient.execute(httpGet)) {
                    HttpEntity respEntity = getHttpEntity(resp);
                    if (respEntity == null) {
                        throw new IOException(String.format("Failed to flush data to Doris, Error " +
                                "could not get the final state of label[%s].\n", label), null);
                    }
                    Map<String, Object> result = (Map<String, Object>) JSONValue.parse(EntityUtils.toString(respEntity));
                    String labelState = (String) result.get("state");
                    if (null == labelState) {
                        throw new IOException(String.format("Failed to flush data to Doris, Error " +
                                "could not get the final state of label[%s]. response[%s]\n", label, EntityUtils.toString(respEntity)), null);
                    }
                    LOG.info(String.format("Checking label[%s] state[%s]\n", label, labelState));
                    switch (labelState) {
                        case LAEBL_STATE_VISIBLE:
                        case LAEBL_STATE_COMMITTED:
                            return;
                        case RESULT_LABEL_PREPARE:
                            continue;
                        case RESULT_LABEL_ABORTED:
                            throw new DorisWriterExcetion(String.format("Failed to flush data to Doris, Error " +
                                    "label[%s] state[%s]\n", label, labelState), null, true);
                        case RESULT_LABEL_UNKNOWN:
                        default:
                            throw new IOException(String.format("Failed to flush data to Doris, Error " +
                                    "label[%s] state[%s]\n", label, labelState), null);
                    }
                }
            }
        }
    }

    // 根据格式将数据行拼接成字节数组
    private byte[] addRows(List<byte[]> rows, int totalBytes) {
        if (Keys.StreamLoadFormat.CSV.equals(options.getStreamLoadFormat())) {
            Map<String, Object> props = (options.getLoadProps() == null ? new HashMap<>() : options.getLoadProps());
            byte[] lineDelimiter = DelimiterParser.parse((String) props.get("line_delimiter"), "\n").getBytes(StandardCharsets.UTF_8);
            ByteBuffer bos = ByteBuffer.allocate(totalBytes + rows.size() * lineDelimiter.length);
            for (byte[] row : rows) {
                bos.put(row);
                bos.put(lineDelimiter);
            }
            return bos.array();
        }

        if (Keys.StreamLoadFormat.JSON.equals(options.getStreamLoadFormat())) {
            ByteBuffer bos = ByteBuffer.allocate(totalBytes + (rows.isEmpty() ? 2 : rows.size() + 1));
            bos.put("[".getBytes(StandardCharsets.UTF_8));
            byte[] jsonDelimiter = ",".getBytes(StandardCharsets.UTF_8);
            boolean isFirstElement = true;
            for (byte[] row : rows) {
                if (!isFirstElement) {
                    bos.put(jsonDelimiter);
                }
                bos.put(row);
                isFirstElement = false;
            }
            bos.put("]".getBytes(StandardCharsets.UTF_8));
            return bos.array();
        }
        throw new RuntimeException("Failed to join rows data, unsupported `format` from stream load properties:");
    }

private Map<String, Object> put(String loadUrl, String label, byte[] data) throws IOException {
        RequestConfig requestConfig = RequestConfig.custom()
                .setSocketTimeout(120 * 1000)
                .setConnectTimeout(120 * 1000)
                .setConnectionRequestTimeout(120 * 1000)
                .build();
        try (CloseableHttpClient httpclient = HttpClientBuilder.create()
                .setDefaultRequestConfig(requestConfig)
                .setRedirectStrategy(new DefaultRedirectStrategy())
                .build()) {
            HttpPut httpPut = new HttpPut(loadUrl);
            httpPut.setHeader(HttpHeaders.CONTENT_TYPE, "application/octet-stream");
            httpPut.setHeader("Authorization", getBasicAuthHeader(options.getUsername(), options.getPassword()));
            httpPut.setEntity(new ByteArrayEntity(data));
            try (CloseableHttpResponse resp = httpclient.execute(httpPut)) {
                HttpEntity respEntity = getHttpEntity(resp);
                if (respEntity == null) {
                    throw new IOException("Failed to flush data to Doris, Error could not get the response entity.");
                }
                return (Map<String, Object>) JSONValue.parse(EntityUtils.toString(respEntity));
            }
        }
    }

    // 构造 HTTP 请求中的基本认证头部
    private String getBasicAuthHeader(String username, String password) {
        String credentials = username + ":" + password;
        byte[] credentialsBytes = credentials.getBytes(StandardCharsets.UTF_8);
        String base64Credentials = Base64.encodeBase64String(credentialsBytes);
        return "Basic " + base64Credentials;
    }

    // 从 HTTP 响应中获取实体内容
    private HttpEntity getHttpEntity(CloseableHttpResponse response) {
        if (response != null) {
            return response.getEntity();
        }
        return null;
    }

    // 获取用于加载数据的主机地址
    private String getLoadHost() {
        List<String> hosts = options.getDorisStreamLoadUrls();
        for (String host : hosts) {
            try {
                HttpURLConnection connection = (HttpURLConnection) new URL(host).openConnection();
                connection.setRequestMethod("HEAD");
                int responseCode = connection.getResponseCode();
                if (responseCode == HttpURLConnection.HTTP_OK) {
                    return host;
                }
            } catch (IOException e) {
                LOG.warn("Failed to connect to host: {}", host);
            }
        }
        return null;
    }
}

作者 east

doris 8月 25,2023

DataX DorisWriter 插件DorisWriterManager类详细解读

DorisWriterManager 的类，用于将数据写入到 Doris 中。以下是代码的具体作用和功能解释：

导入必要的包和类： 代码开头导入了所需的包和类，包括日志记录、线程池、字符编码和其他相关工具类。
类成员变量定义： 下面是一些类的成员变量定义，这些变量在类的不同方法中使用：
- LOG: 用于记录日志的 Logger 对象。
- visitor: DorisStreamLoadObserver 类的实例，用于处理数据写入 Doris 的观察者。
- options: Keys 类的实例，包含了一些配置选项。
- buffer: 存储待写入 Doris 的数据。
- batchCount: 当前批次中的记录数量。
- batchSize: 当前批次中的数据大小。
- closed: 标志位，表示是否已关闭写入。
- flushException: 异步刷新数据时可能发生的异常。
- flushQueue: 用于异步刷新数据的队列。
- scheduler: 用于定期刷新数据的调度器。
- scheduledFuture: 用于取消定时任务的句柄。
构造函数 DorisWriterManager： 构造函数接受一个 Keys 对象作为参数，设置了初始化的配置信息，并初始化了 visitor 和 flushQueue。接着，它调用 startScheduler() 启动定期刷新任务，以及 startAsyncFlushing() 启动异步刷新线程。
startScheduler() 方法： 此方法负责启动定时刷新任务。它首先调用 stopScheduler() 停止之前的定时任务。然后，创建一个单线程的调度器（scheduler），并设置一个定时任务，定期触发数据刷新操作。在定时任务内部，它会检查是否关闭了写入操作，然后根据配置信息进行数据刷新。如果当前批次为空，重新启动定时任务，确保数据持续刷新。
stopScheduler() 方法： 此方法用于停止定时任务。它会取消之前的定时任务并关闭调度器。
writeRecord(String record) 方法： 该方法用于将记录写入缓冲区。它首先调用 checkFlushException() 方法检查是否存在刷新异常。然后，将记录转换成字节数组并添加到缓冲区中，同时更新批次计数和数据大小。如果当前批次的记录数量或数据大小超过了阈值，就会触发数据刷新。
flush(String label, boolean waitUntilDone) 方法： 此方法用于手动触发数据刷新操作。它首先检查是否存在刷新异常，然后根据当前批次的情况决定是否执行刷新。如果当前批次为空，且 waitUntilDone 为真，它会等待之前的异步刷新操作完成。否则，它将当前批次的数据放入刷新队列，并根据 waitUntilDone 参数决定是否等待刷新操作完成。
close() 方法： 此方法用于关闭 DorisWriterManager。它首先检查是否已经关闭，然后触发一次最终的数据刷新操作。如果当前批次有数据，会记录相应日志。最后，它检查是否有刷新异常并抛出相应异常。
createBatchLabel() 方法： 此方法用于创建批次标签，用于标识一批数据。它根据配置的前缀和随机 UUID 生成标签。
startAsyncFlushing() 方法： 此方法启动一个异步刷新线程。线程会循环调用 asyncFlush() 方法，将数据异步刷新到 Doris 中。
waitAsyncFlushingDone() 方法： 该方法用于等待之前的异步刷新操作完成。它向刷新队列添加空的 WriterTuple，以确保之前的刷新操作完成。然后，它检查是否存在刷新异常。
asyncFlush() 方法： 此方法用于异步刷新数据到 Doris。它从刷新队列中取出 WriterTuple，然后根据批次的标签执行数据刷新操作。如果发生异常，它会尝试多次，直到达到最大重试次数。如果需要重新创建批次标签，则生成新的标签。重试之间会休眠一段时间。成功后，重新启动定时任务。
checkFlushException() 方法： 此方法用于检查是否存在刷新异常，如果存在则抛出异常。

这个 DorisWriterManager 类的目的是管理数据写入到 Doris 数据库的操作。它通过定时任务和异步刷新线程来控制数据的批量写入，同时处理异常情况，确保数据的稳定写入。

添加详细注释代码如下:

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DorisWriterManager {

    private static final Logger LOG = LoggerFactory.getLogger(DorisWriterManager.class);

    private final DorisStreamLoadObserver visitor;
    private final Keys options;
    private final List<byte[]> buffer = new ArrayList<>(); // 缓冲区，用于存储待写入 Doris 的数据
    private int batchCount = 0; // 当前批次中的记录数量
    private long batchSize = 0; // 当前批次中的数据大小
    private volatile boolean closed = false; // 标志位，表示是否已关闭
    private volatile Exception flushException; // 异步刷新数据时可能发生的异常
    private final LinkedBlockingDeque<WriterTuple> flushQueue; // 用于异步刷新数据的队列
    private ScheduledExecutorService scheduler; // 用于定期刷新数据的调度器
    private ScheduledFuture<?> scheduledFuture;

    public DorisWriterManager(Keys options) {
        this.options = options;
        this.visitor = new DorisStreamLoadObserver(options);
        flushQueue = new LinkedBlockingDeque<>(options.getFlushQueueLength());
        this.startScheduler(); // 启动定期刷新调度器
        this.startAsyncFlushing(); // 启动异步刷新线程
    }

    // 启动定期刷新调度器
    public void startScheduler() {
        stopScheduler(); // 停止之前的调度器
        this.scheduler = Executors.newScheduledThreadPool(1, new BasicThreadFactory.Builder()
                .namingPattern("Doris-interval-flush").daemon(true).build());
        this.scheduledFuture = this.scheduler.schedule(() -> {
            synchronized (DorisWriterManager.this) {
                if (!closed) {
                    try {
                        String label = createBatchLabel();
                        LOG.info(String.format("Doris interval Sinking triggered: label[%s].", label));
                        if (batchCount == 0) {
                            startScheduler(); // 如果当前批次为空，重新启动定时任务
                        }
                        flush(label, false);
                    } catch (Exception e) {
                        flushException = e;
                    }
                }
            }
        }, options.getFlushInterval(), TimeUnit.MILLISECONDS);
    }

    // 停止定期刷新调度器
    public void stopScheduler() {
        if (this.scheduledFuture != null) {
            scheduledFuture.cancel(false);
            this.scheduler.shutdown();
        }
    }

    // 写入一条记录到缓冲区
    public final synchronized void writeRecord(String record) throws IOException {
        checkFlushException(); // 检查是否有刷新异常
        try {
            byte[] bts = record.getBytes(StandardCharsets.UTF_8);
            buffer.add(bts);
            batchCount++;
            batchSize += bts.length;
            if (batchCount >= options.getBatchRows() || batchSize >= options.getBatchSize()) {
                String label = createBatchLabel();
                LOG.debug(String.format("Doris buffer Sinking triggered: rows[%d] label[%s].", batchCount, label));
                flush(label, false); // 当记录数量或数据大小超过阈值时触发刷新
            }
        } catch (Exception e) {
            throw new IOException("Writing records to Doris failed.", e);
        }
    }

    // 手动触发刷新缓冲区的数据
    public synchronized void flush(String label, boolean waitUntilDone) throws Exception {
        checkFlushException(); // 检查是否有刷新异常
        if (batchCount == 0) {
            if (waitUntilDone) {
                waitAsyncFlushingDone(); // 如果当前批次为空，等待之前的刷新操作完成
            }
            return;
        }
        flushQueue.put(new WriterTuple(label, batchSize, new ArrayList<>(buffer))); // 将数据放入刷新队列
        if (waitUntilDone) {
            waitAsyncFlushingDone(); // 等待刷新操作完成
        }
        buffer.clear();
        batchCount = 0;
        batchSize = 0;
    }

    // 关闭 DorisWriterManager，触发最后一次刷新操作
    public synchronized void close() {
        if (!closed) {
            closed = true;
            try {
                String label = createBatchLabel();
                if (batchCount > 0) LOG.debug(String.format("Doris Sink is about to close: label[%s].", label));
                flush(label, true); // 关闭时触发刷新操作
            } catch (Exception e) {
                throw new RuntimeException("Writing records to Doris failed.", e);
            }
        }
        checkFlushException();
    }

    // 创建批次标签，通常用于标识一批数据
    public String createBatchLabel() {
        StringBuilder sb = new StringBuilder();
        if (!Strings.isNullOrEmpty(options.getLabelPrefix())) {
            sb.append(options.getLabelPrefix());
        }
        return sb.append(UUID.randomUUID().toString()).toString();
    }

    // 启动异步刷新线程
    private void startAsyncFlushing() {
        Thread flushThread = new Thread(new Runnable() {
            public void run() {
                while (true) {
                    try {
                        asyncFlush(); // 异步刷新数据
                    } catch (Exception e) {
                        flushException = e;
                    }
                }
            }
        });
        flushThread.setDaemon(true);
        flushThread.start();
    }

    // 等待之前的刷新操作完成
    private void waitAsyncFlushingDone() throws InterruptedException {
        for (int i = 0; i <= options.getFlushQueueLength(); i++) {
            flushQueue.put(new WriterTuple("", 0L, null));
        }
        checkFlushException();
    }

    // 异步刷新数据到 Doris
    private void asyncFlush() throws Exception {
        WriterTuple flushData = flushQueue.take();
        if (Strings.isNullOrEmpty(flushData.getLabel())) {
            return;
        }
        stopScheduler(); // 停止定时任务
        LOG.debug(String.format("Async stream load: rows[%d] bytes[%d] label[%s].", flushData.getRows().size(), flushData.getBytes(), flushData.getLabel()));
        for (int i = 0; i <= options.getMaxRetries(); i++) {
            try {
                // 利用 DorisStreamLoadObserver 进行数据刷新
                visitor.streamLoad(flushData);
                LOG.info(String.format("Async stream load finished: label[%s].", flushData.getLabel()));
                startScheduler(); // 
     break;
            } catch (Exception e) {
                LOG.warn("Failed to flush batch data to Doris, retry times = {}", i, e);
                if (i >= options.getMaxRetries()) {
                    throw new IOException(e);
                }
                if (e instanceof DorisWriterExcetion && (( DorisWriterExcetion )e).needReCreateLabel()) {
                    String newLabel = createBatchLabel();
                    LOG.warn(String.format("Batch label changed from [%s] to [%s]", flushData.getLabel(), newLabel));
                    flushData.setLabel(newLabel);
                }
                try {
                    Thread.sleep(1000l * Math.min(i + 1, 10));
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                    throw new IOException("Unable to flush, interrupted while doing another attempt", e);
                }
            }
        }
    }

    private void checkFlushException() {
        if (flushException != null) {
            throw new RuntimeException("Writing records to Doris failed.", flushException);
        }
    }
}

作者 east

doris 7月 18,2023

Spark写数据到Doris报错node and exceeded the max retry times

用spark dataframe向doris写数据时，报下面错误：

Failed to load data on BE: http://192.168.50.10:18040/api/mydb/dwd_virtual_table/_stream_load? node and exceeded the max retry times.

发现表没写入成功。刚开始很困惑，后来发现是 dataFrame中的字段和目标表不一致。

这种提示很不友好，有没有更好方式提示，方法是有的，可以用jdbc写入，发现错误时可以看到具体的提示。代码参考如下：

def writeByJDBC(dataframe: DataFrame, dorisTable: String): Unit = {
    dataframe.write.format("jdbc")
      .mode(SaveMode.Append)
      .option("driver", "com.mysql.jdbc.Driver")
      .option("url", "jdbc:mysql://" + DORIS_HOST + ":9030/" +DATABASE_NAME + "?rewriteBatchedStatements=false")
      .option("batchsize", "" + WRITE_BATCH_SIZE)
      .option("user", DORIS_USER)
      .option("password", DORIS_PASSWORD)
      .option("isolationLevel", "NONE")
      //  .option("doris.write.fields","case_id,defendant_name,finance_name,mediation_name,mediator_name,dt")
      .option("dbtable", dorisTable)
      .save()
  }

不过这种方式还是没有Spark Doris Connector的方式写效率高，可以用上面jdbc方式调试，没问题再切换 Spark Doris Connector 方式：

def writeByDoris(dataframe: DataFrame, dorisTable: String): Unit = {
dataframe.write.format(“doris”)
.option(“doris.table.identifier”, dorisTable)
.option(“doris.fenodes”, DORIS_HOST + “:” + DORIS_FE_HTTP_PORT)
.option(“user”, DORIS_USER)
.option(“password”, DORIS_PASSWORD)
.option(“sink.batch.size”, WRITE_BATCH_SIZE)
.option(“sink.max-retries”, 3)
.option(“doris.request.retries”, 6)
.option(“doris.request.retries”, 100)
.option(“doris.request.connect.timeout.ms”, 60000)
.save()
}

关注公众号“大模型全栈程序员”回复“大数据面试”获取800页左右大数据面试宝典，回复“大数据”获取多本大数据电子书

作者 east

doris 6月 13,2023

Spark 写入数据到doris异常

用spark写入到doris，由于服务器配置不高， sink.batch.size 设置过大时，超出服务器内存限制，doris be端直接闪崩。如果 sink.batch.size 设置过小，则出现写入次数太频繁无法写入。调整 write 的
sink.batch.size 参数大小，逐步增加 batchsize 大小，以达到更好的写入性能与内存占用的平衡。例如，可以逐渐将 batchsize 参数的值从 200 调整到 500，1000，2000，以找到最好的性能与内存占用平衡点。刚开始设置10000时服务器be端闪崩，后来设置200时又写到中途报错，最后找到2000这个平衡点。

def writeDoris(dataframe: DataFrame, dorisTable: String): Unit = {
dataframe.write.format(“doris”)
.option(“doris.table.identifier”, dorisTable)
.option(“doris.fenodes”, DORIS_HOST + “:” + DORIS_FE_HTTP_PORT)
.option(“user”, DORIS_USER)
.option(“password”, DORIS_PASSWORD)
.option(“sink.batch.size”, 2000)
.save()
}

关注公众号“大模型全栈程序员”回复“大数据面试”获取800页左右大数据面试宝典，回复“大数据”获取多本大数据电子书

作者 east

doris 6月 8,2023

spark运行写入doris的任务报错park.DorisStreamLoad: Doris BE’s response cannot map to schema

在CDH的集群上跑spark任务，运行报错：
ERROR spark.DorisStreamLoad: Doris BE’s response cannot map to schema. res: “Access denied for user ‘default_cluster:hdfs@10.0.10.101’ (using password: YES)”
org.apache.doris.shaded.com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot construct instance of org.apache.doris.spark.rest.models.BackendV2 (although at least one Creator exists): no String-argument constructor/factory method to deserialize from String value (‘Access denied for user ‘default_cluster:hdfs@10.0.10.101′ (using password: YES)’)
at [Source: (String)””Access denied for user ‘default_cluster:hdfs@10.0.10.101’

查了一下，连接doris的用户名和密码没错。后来发现，spark-submit的参数–conf spark.user=hdfs 影响了，把这个参数去掉就正常了。

关注公众号“大模型全栈程序员”回复“大数据面试”获取800页左右大数据面试宝典，回复“大数据”获取多本大数据电子书

作者 east

doris 6月 1,2023

spark jdbc模式写入异常

用spark jdbc模式写入doris报下面错误：

java.sql.BatchUpdateException: Insert has filtered data in strict mode, tracking_url=http://192.168.1.40:8040/api/_load_error_log?file=__shard_49/error_log_insert_stmt_8384ff7a733e4ce5-bc24c049a085fdd2_8384ff7a733e4ce5_bc24c049a085fdd2
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
	at com.mysql.cj.util.Util.handleNewInstance(Util.java:192)
	at com.mysql.cj.util.Util.getInstance(Util.java:167)
	at com.mysql.cj.util.Util.getInstance(Util.java:174)
	at com.mysql.cj.jdbc.exceptions.SQLError.createBatchUpdateException(SQLError.java:224)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchedInserts(ClientPreparedStatement.java:755)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchInternal(ClientPreparedStatement.java:426)
	at com.mysql.cj.jdbc.StatementImpl.executeBatch(StatementImpl.java:800)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.savePartition(JdbcUtils.scala:667)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$saveTable$1.apply(JdbcUtils.scala:834)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$saveTable$1.apply(JdbcUtils.scala:834)
	at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$28.apply(RDD.scala:935)
	at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$28.apply(RDD.scala:935)
	at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2121)
	at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2121)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
	at org.apache.spark.scheduler.Task.run(Task.scala:121)
	at org.apache.spark.executor.Executor$TaskRunner$$anonfun$11.apply(Executor.scala:407)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1408)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:413)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.sql.SQLException: Insert has filtered data in strict mode, tracking_url=http://10.0.80.54:8040/api/_load_error_log?file=__shard_49/error_log_insert_stmt_8384ff7a733e4ce5-bc24c049a085fdd2_8384ff7a733e4ce5_bc24c049a085fdd2
	at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:129)
	at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:122)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeInternal(ClientPreparedStatement.java:953)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeUpdateInternal(ClientPreparedStatement.java:1098)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeUpdateInternal(ClientPreparedStatement.java:1046)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeLargeUpdate(ClientPreparedStatement.java:1371)
	at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchedInserts(ClientPreparedStatement.java:746)
	... 17 more

点开上面的=http://192.168.1.40:8040/api/_load_error_log?file=__shard_49/error_log_insert_stmt_8384ff7a733e4ce5-bc24c049a085fdd2_8384ff7a733e4ce5_bc24c049a085fdd2，发现报下面的错误：

Reason: no partition for this tuple. tuple=

查看要写入doris对应的表，这个表是动态分区的，发现这个表没有插入数据的分区。后来匹配对应的doris表动态分区后，插入数据果然正常了。

关注公众号“大模型全栈程序员”回复“大数据面试”获取800页左右大数据面试宝典，回复“大数据”获取多本大数据电子书

作者 east

doris 5月 26,2023

datax写入doris报错：Writing records to Doris failed

用datax同步数据到doris，报下面错误：

java.lang.RuntimeException: Writing records to Doris failed.
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.close(DorisWriterManager.java:113)
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriter$Task.post(DorisWriter.java:150)
 at com.alibaba.datax.core.taskgroup.runner.WriterRunner.run(WriterRunner.java:65)
 at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.RuntimeException: Writing records to Doris failed.
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.checkFlushException(DorisWriterManager.java:189)
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.waitAsyncFlushingDone(DorisWriterManager.java:150)
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.flush(DorisWriterManager.java:98)
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.close(DorisWriterManager.java:111)
 ... 3 more
Caused by: java.io.IOException: java.io.IOException: Failed to flush data to Doris.
{"TxnId":-1,"Label":"datax_doris_writer_fed9613c-8d95-4284-a9a2-949985cf3f8d","TwoPhaseCommit":"false","Status":"Fail","Message":"errCode = 7, detailMessage = unknown table, tableName=mediation","NumberTotalRows":0,"NumberLoadedRows":0,"NumberFilteredRows":0,"NumberUnselectedRows":0,"LoadBytes":0,"LoadTimeMs":0,"BeginTxnTimeMs":0,"StreamLoadPutTimeMs":0,"ReadDataTimeMs":0,"WriteDataTimeMs":0,"CommitAndPublishTimeMs":0}
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.asyncFlush(DorisWriterManager.java:170)
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager.access$000(DorisWriterManager.java:19)
 at com.alibaba.datax.plugin.writer.doriswriter.DorisWriterManager$1.run(DorisWriterManager.java:134)
 ... 1 more
Caused by: java.io.IOException: Failed to flush data to Doris.

在这种情况下，错误是由于未知表引起的。这可能是由许多因素引起的，例如：

表不存在。
表不可访问。
表配置不正确。
要修复错误，您需要确定错误的原因并采取措施进行纠正。如果表不存在，则需要创建表。如果无法访问表，则需要授予用户对表的写入权限。如果表配置不正确，则需要正确配置它。

一旦您纠正了错误的原因，就应该能够无问题地将数据写入Doris中。

以下是一些解决此错误的附加提示：

检查Doris日志以获取更多有关错误的信息。
尝试使用其他工具（例如Doris CLI或Doris API）将数据写入Doris。
如果仍然遇到问题，您可以联系Doris支持团队寻求帮助。

关注公众号“大模型全栈程序员”回复“大数据面试”获取800页左右大数据面试宝典，回复“大数据”获取多本大数据电子书

作者 east

doris 5月 19,2023

doris批量导出表结构java版本实现

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileWriter;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

public class DorisDump {

    private static final String JDBC_DRIVER = "com.mysql.cj.jdbc.Driver";
    private static final String DB_URL = "jdbc:mysql://localhost:9030/database";
    private static final String USER = "user";
    private static final String PASSWORD = "password";

    public static void main(String[] args) throws SQLException, IOException {
        // Parse command line arguments
        Args args = new Args(args);

        // Connect to the database
        Connection connection = DriverManager.getConnection(DB_URL, USER, PASSWORD);

        // Create a statement
        Statement statement = connection.createStatement();

        // Get the list of tables
        ResultSet tables = statement.executeQuery("show tables;");

        // Create a list to store the create table statements
        List<String> createTableStatements = new ArrayList<>();

        // Iterate over the tables
        while (tables.next()) {
            // Get the name of the table
            String tableName = tables.getString(1);

            // Get the create table statement for the table
            ResultSet createTableStatement = statement.executeQuery("show create table " + tableName + ";");

            // Add the create table statement to the list
            createTableStatements.add(createTableStatement.getString(1));
        }

        // Write the create table statements to a file
        File file = new File("table.txt");
        try (FileWriter fileWriter = new FileWriter(file)) {
            for (String createTableStatement : createTableStatements) {
                fileWriter.write(createTableStatement + "\n");
            }
        }

        // Close the connection
        connection.close();
    }

    private static class Args {

        private String host;
        private int port;
        private String user;
        private String password;
        private String database;

        public Args(String[] args) {
            for (int i = 0; i < args.length; i++) {
                if (args[i].equals("-h")) {
                    host = args[i + 1];
                } else if (args[i].equals("-P")) {
                    port = Integer.parseInt(args[i + 1]);
                } else if (args[i].equals("-u")) {
                    user = args[i + 1];
                } else if (args[i].equals("-p")) {
                    password = args[i + 1];
                } else if (args[i].equals("-d")) {
                    database = args[i + 1];
                }
            }
        }

        public String getHost() {
            return host;
        }

        public int getPort() {
            return port;
        }

        public String getUser() {
            return user;
        }

        public String getPassword() {
            return password;
        }

        public String getDatabase() {
            return database;
        }
    }
}

关注公众号“大模型全栈程序员”回复“大数据面试”获取800页左右大数据面试宝典，回复“大数据”获取多本大数据电子书

作者 east

doris, Flink 5月 14,2023

要对接mysql数据到doris，对比一下用flink cdc和datax对接数据性能和简易性

用flink cdc和datax对接mysql数据到doris的性能和简易性有以下几点对比：

flink cdc可以实现实时的数据同步，而datax只能实现批量的数据同步，因此flink cdc在性能上有优势，可以满足低延迟的需求。
flink cdc可以利用flink的容错机制和检查点机制，保证数据的一致性和精确一次语义，而datax没有这样的机制，可能会出现数据丢失或重复的情况。
flink cdc可以利用flink的流批一体能力，对接doris和其他数据源，进行联合分析和处理，而datax只能单纯地进行数据传输，没有分析和处理的能力。
flink cdc需要使用flink sql或者flink api编写作业，并配置好相关的参数和选项，而datax只需要编写json格式的配置文件，并运行相应的命令，因此datax在简易性上有优势，更适合不熟悉flink的用户。
flink cdc需要部署和运行一个flink集群，并监控和维护作业的状态，而datax可以直接在本地或者服务器上运行，不需要额外的资源和管理，因此datax在成本上有优势，更适合小规模或临时的数据同步任务。

综上所述，flink cdc和datax都有各自的优缺点，具体选择哪种方案要根据你的具体需求和场景来决定。

作者 east

doris, Flink 5月 11,2023

同步 MySQL 数据到 Doris 的常用方案

使用Flink CDC

优点：

实时同步，可支持增量同步和全量同步
可以按需调整并发度和流水线等参数，可弹性伸缩
高可靠性和灵活性，能够处理多种场景下的同步需求

缺点：

实现和部署比较复杂，需要掌握 Flink 的相关知识
对系统资源和性能要求较高，需要有一定的硬件配置
需要进行一定的性能测试和优化，才能支持实际业务量

2.使用Doris MySQL Proxy

优点：

部署和使用比较简单，不需要修改 MySQL 应用程序
对 MySQL 应用程序和数据库系统无侵入
可以保证数据的一致性和可靠性

缺点：

过程中需要对数据库 Proxy 进行自定义集成
不支持多库多表的同步，只能同步一个 MySQL 数据库
可能存在数据丢失或者漏读的问题

3.使用Canal

优点：

存在简单，部署方便
支持 MySQL 的多种版本
可用于实时同步 MySQL 数据
对系统资源和性能需求较低，适合小数据量同步

缺点：

不支持多库多表同步
可能存在数据丢失或者漏读的问题
官方不再维护

作者 east

分类归档doris