Doris Stream Load 实战：Java 服务端实时写入与性能调优

1. 为什么我最终选了 Stream Load 而不是其他导入方式做数据平台这几年Doris 的导入方式我基本都趟过一遍。Insert Into 适合小批量低频写入Broker Load 适合 HDFS 上的大文件Routine Load 适合 Kafka 常驻消费而 Stream Load 是我在服务端实时写入场景下最常用的一种。它的本质是通过 HTTP 协议把本地文件或内存数据推给 Doris FEFE 再协调 BE 完成写入整个过程同步返回结果省去了中间存储的依赖。这次的需求背景很典型上游是一个 Java 写的业务服务每产生一批订单数据就要尽快落到 Doris 里供报表查询。数据量不算大单批几千到几万行但要求延迟低、失败可感知、能重试。用 Stream Load 直接通过 Java 的 HTTP 客户端调用 FE 接口是最短路径。很多人一上来会纠结要不要用 Flink Connector 或者 Spark其实对于这种“业务系统直接写”的场景引入计算引擎反而是负担。Stream Load 的接口足够简单一个 HTTP PUT 请求就能搞定Java 里用 HttpClient 或者 OkHttp 都能实现不需要额外依赖 Doris 的 SDK。提示Stream Load 是同步导入单次请求的数据量建议控制在 1GB 以内超大文件还是走 Broker Load 更稳妥。1.1 Stream Load 的工作机制拆解理解机制才能用好它。当你发起一个 Stream Load 请求时FE 收到后并不会自己存数据它做的是事务协调先根据表的分桶信息把数据按分区和 tablet 切分然后分发给对应的 BE。BE 收到数据后写入本地存储完成后回报 FEFE 再统一返回给客户端。这里有个关键点FE 只负责协调不落数据。所以你在 Java 代码里配置的地址是 FE 的 HTTP 端口默认 8030而不是 BE 的。很多人第一次调的时候填了 BE 地址结果一直报连接错误就是没搞清楚这个分工。另一个要理解的是两阶段提交。Stream Load 在 BE 侧先写临时文件等所有 BE 都写完并且 FE 确认事务可以提交后才真正生效。这意味着如果中途失败数据不会半写进去保证了原子性。这个特性对业务系统很重要——要么全成功要么全失败不会出现“写了一半”的脏数据。1.2 和其他导入方式的对比选型我把几种常见方式列个表方便你对照自己的场景选导入方式数据源同步/异步适用场景我的使用频率Stream Load本地文件/内存同步服务端实时写入最高Broker LoadHDFS/对象存储异步大批量离线导入中Routine LoadKafka异步常驻流式持续消费中Insert IntoSQL 客户端同步小批量测试低选 Stream Load 的核心理由就三条延迟低请求返回即写入完成、无外部依赖不需要 Kafka 或 HDFS、Java 集成简单纯 HTTP。如果你的场景是业务系统产生数据后要立刻可查Stream Load 基本是最优解。2. Java 调用 FE 接口的核心细节与参数配置真正动手写代码之前有几个参数必须搞清楚否则调通了也是“能跑但跑不好”的状态。Stream Load 的请求头里有一堆配置项我挑最关键的几个讲。2.1 认证与地址配置Doris 默认开启 HTTP 基础认证所以请求头里要带Authorization。Java 里用 Base64 编码用户名密码即可String auth Base64.getEncoder().encodeToString( (username : password).getBytes(StandardCharsets.UTF_8)); request.setHeader(Authorization, Basic auth);地址格式是http://FE_HOST:8030/api/{db}/{table}/_stream_load。注意路径里的_stream_load是固定后缀不能少。如果 FE 是多节点建议配一个负载均衡地址或者自己做个简单的轮询避免单点压力。2.2 关键请求头参数详解这些参数直接决定导入行为和性能我逐个说明label本次导入的唯一标识。同一个 label 重复提交Doris 会认为是同一次导入直接返回上次结果。这个机制天然支持幂等重试——网络超时后你拿同样的 label 重发不会产生重复数据。我一般用业务名_表名_时间戳_UUID的格式生成。column_separator列分隔符默认是\t。如果你的数据是 CSV要改成,。支持多字符分隔符比如||。line_delimiter行分隔符默认\n。一般不用改除非数据里有特殊换行。format数据格式支持csv、json、parquet等。我用 CSV 最多JSON 次之。max_filter_ratio允许的过滤比例默认 0。意思是数据里如果有脏行比如格式不对超过这个比例就整个导入失败。生产环境我一般设成 0.1容忍少量脏数据避免因为一行问题导致整批失败。strict_mode严格模式开启后会对字段类型做严格校验。建议开启能提前发现数据质量问题。timeout超时时间默认 600 秒。大批量导入要适当调大。exec_mem_limit单个导入任务的内存限制默认 2GB。数据量大时调大避免 OOM。注意max_filter_ratio设太大有风险脏数据会被静默丢弃。我一般配合导入后的行数校验来用如果实际写入行数和发送行数差异超过预期就告警排查。2.3 数据体构造的两种方式数据体就是你要导入的内容放在 HTTP 请求的 body 里。两种常见构造方式方式一直接拼字符串。适合数据量小、字段少的情况。用 StringBuilder 逐行拼接注意转义分隔符和换行符。方式二流式写入。数据量大时用InputStream或者ByteArrayInputStream避免在内存里拼一个巨大的字符串。Java HttpClient 支持BodyPublishers.ofInputStream()可以边读边发。我实测下来单批 5 万行以内用字符串拼接完全没问题超过这个量建议走流式否则 GC 压力会很明显。3. 完整实操从零写一个可复用的导入工具类光讲参数不够直接上代码。下面这个工具类是我在多个项目里迭代出来的封装了认证、参数配置、异常处理和结果解析你可以直接拿去改。3.1 依赖准备我用的是 Java 11 自带的java.net.http.HttpClient不需要额外引依赖。如果你还在 Java 8换成 OkHttp 或 Apache HttpClient 即可逻辑一样。!-- 如果用 OkHttp -- dependency groupIdcom.squareup.okhttp3/groupId artifactIdokhttp/artifactId version4.12.0/version /dependency3.2 核心导入方法实现public class DorisStreamLoader { private final String feHost; private final int fePort; private final String username; private final String password; private final HttpClient httpClient; public DorisStreamLoader(String feHost, int fePort, String username, String password) { this.feHost feHost; this.fePort fePort; this.username username; this.password password; this.httpClient HttpClient.newBuilder() .connectTimeout(Duration.ofSeconds(10)) .build(); } public LoadResult load(String db, String table, String data, String label) throws IOException, InterruptedException { String url String.format(http://%s:%d/api/%s/%s/_stream_load, feHost, fePort, db, table); String auth Base64.getEncoder().encodeToString( (username : password).getBytes(StandardCharsets.UTF_8)); HttpRequest request HttpRequest.newBuilder() .uri(URI.create(url)) .header(Authorization, Basic auth) .header(label, label) .header(column_separator, ,) .header(format, csv) .header(strict_mode, true) .header(max_filter_ratio, 0.1) .header(timeout, 300) .PUT(HttpRequest.BodyPublishers.ofString(data, StandardCharsets.UTF_8)) .build(); HttpResponseString response httpClient.send(request, HttpResponse.BodyHandlers.ofString()); return parseResult(response.body()); } private LoadResult parseResult(String body) { // 解析 JSON 响应提取 Status、LoadedRows、ErrorURL 等字段 // 这里用简单的字符串匹配生产环境建议用 Jackson LoadResult result new LoadResult(); result.setSuccess(body.contains(\Status\: \Success\)); // ... 其他字段解析 return result; } }3.3 调用示例与数据构造public class Demo { public static void main(String[] args) throws Exception { DorisStreamLoader loader new DorisStreamLoader( 192.168.1.100, 8030, root, ); StringBuilder sb new StringBuilder(); sb.append(1001,张三,2024-01-15,99.50\n); sb.append(1002,李四,2024-01-15,128.00\n); sb.append(1003,王五,2024-01-15,56.80\n); String label order_import_ System.currentTimeMillis(); LoadResult result loader.load(sales_db, orders, sb.toString(), label); if (result.isSuccess()) { System.out.println(导入成功行数 result.getLoadedRows()); } else { System.err.println(导入失败 result.getErrorMsg()); System.err.println(错误详情 result.getErrorUrl()); } } }这段代码跑通后你会看到 Doris 返回一个 JSON里面有Status、LoadedRows、FilteredRows、ErrorURL等字段。ErrorURL特别有用导入失败时它会指向一个文件里面记录了每一行失败的原因排查问题全靠它。3.4 批量导入的性能调优单次导入几千行没问题但如果要导入几十万行得注意几点分批发送。我一般按 5 万行一批切分多批并行发送。Doris 对并发导入的支持很好但要注意 label 不能重复每批用不同的 label。控制并发数。并发太高会给 BE 造成压力我实测 4 到 8 个并发是比较舒服的区间具体看 BE 的配置。压缩传输。请求头加compress_type: gzipbody 用 GZIP 压缩后再发能显著减少网络传输时间。数据量大时效果很明显。byte[] compressed gzipCompress(data.getBytes(StandardCharsets.UTF_8)); request.header(compress_type, gzip) .PUT(HttpRequest.BodyPublishers.ofByteArray(compressed));4. 踩过的坑与常见问题排查这部分是我最想分享的因为文档里不会写但实际用起来一定会遇到。4.1 导入报错排查速查表错误现象可能原因解决方法401 Unauthorized用户名密码错误或未带认证头检查 Authorization 头404 Not Found库名表名错误或路径拼错确认_stream_load后缀400 Bad Request参数格式错误检查 column_separator 等Status: Fail, ErrorURL 有值数据格式问题下载 ErrorURL 查看具体行连接超时FE 地址不通或端口错确认 8030 端口可达导入成功但行数为 0数据为空或全被过滤检查数据内容和 filter_ratio4.2 几个让我印象深刻的坑坑一label 重复导致数据丢失。有次我用固定 label 做测试第二次导入时 Doris 直接返回上次的成功结果数据根本没进去。后来改成时间戳加 UUID 才解决。记住label 是幂等键不是随便填的。坑二分隔符冲突。数据里本身含有逗号我又用逗号做分隔符结果列数对不上。解决办法是换一个数据里不会出现的分隔符比如\u0001或者对数据做转义。坑三时区问题。Doris 的日期字段对时区敏感Java 端如果用的是本地时区写入后查询可能差几个小时。建议统一用 UTC 或者明确指定时区。坑四大批量导入 OOM。一次导入 100 万行Java 端拼字符串直接把内存撑爆。后来改成流式发送加分批问题解决。提示每次导入后一定要检查FilteredRows如果这个值不为 0说明有数据被过滤了得去 ErrorURL 看原因。4.3 生产环境的稳定性建议加超时和重试。网络抖动是常态我一般设 3 次重试每次用相同的 label保证幂等。监控导入延迟和成功率。把每次导入的耗时、行数、状态打点到监控系统出问题能第一时间发现。定期清理 label。Doris 会保留 label 历史时间长了会占元数据空间。可以配置自动清理策略。数据量特别大时考虑攒批。业务系统如果每秒都产生数据不要每秒都发一次导入攒个几秒或几千行再发能大幅降低 FE 压力。5. 这套方案还能怎么扩展Stream Load 跑通之后我后来又做了几个延伸。一个是把导入工具类封装成 Spring Boot Starter业务代码里加个注解就能用。另一个是接入了重试队列导入失败的批次自动进队列后台定时重试。还有一个是在导入前加了数据校验层字段类型、长度、枚举值都先过一遍把脏数据挡在 Doris 外面。如果你也在做类似的数据导入我的建议是先把单批导入跑通再考虑并发和容错。不要一上来就搞复杂的架构Stream Load 本身已经足够简单可靠把参数调对、把异常处理好就能覆盖大部分场景。