★ FEATURED ARTICLE 简介doris-stream-loader 是 doris 官方为 Kettle-Spoon 提供的批量加载插件面向需要将数据从 Kettle 高效抽取至 doris 数据库的 ETL 开发与数据分析人员尤其适合海量数据实时分析场景下的数据同步工作。资源包共 4 个文件以 3 个 jar 与 1 个 xml 为主jar 分别承担插件实现与界面逻辑xml 用于版本声明压缩包整体约 502KB体积轻便、部署简单。目前已有 1544 人学习下载具备一定参考热度。插件要求 Kettle 版本为 9.4.0.0-343解压后放入># 进入插件源码目录编译打包 mvn clean package -DskipTests # 编译产物在 target 目录下形如 doris-stream-loader-xxx.jar # 复制到 Kettle 的 steps 插件目录 cp target/doris-stream-loader-*.jar $KETTLE_HOME/plugins/steps/编译时如果报依赖找不到检查settings.xml里的镜像配置Doris 的部分依赖不在中央仓库。打包完成后重启 Spoon新建转换在输出分类下应该能看到 Doris Stream Loader 步骤。3.2 配置 Stream Load 连接参数拖入步骤后双击配置。核心参数分三组连接、目标、导入选项。连接组里填 FE 的 HTTP 地址和端口注意是 FE 的 8030 端口不是 BE 的 8040。插件会先向 FE 请求重定向拿到实际负责写入的 BE 地址。用户名密码填 Doris 的账号默认 root 空密码在测试环境可用生产必须建独立账号。目标组填数据库名和表名。表名要和 Doris 里的大小写完全一致Doris 默认表名大小写敏感。如果表是分区表不需要额外配置Stream Load 会根据数据里的分区键自动路由。导入选项组是重点。batchSize控制攒批行数默认 10000batchByteSize控制攒批字节数默认 10MBmaxRetries控制失败重试次数默认 3。这三个参数后面会专门讲怎么调。3.3 字段映射与数据格式插件支持 CSV 和 JSON 两种格式。CSV 模式下上游字段按顺序对应表的列可以用columns参数指定列名和顺序。JSON 模式下每个字段名对应 JSON 的 key。字段映射的配置界面和 Kettle 其他步骤类似左边是上游字段右边是目标列。这里有个容易翻车的点Doris 的列类型和 Kettle 的字段类型不是一一对应的。比如 Doris 的DECIMAL(10,2)对应 Kettle 的BigNumber如果上游是String类型插件会尝试转换转换失败整批数据都会被拒。-- 建一张测试表字段类型覆盖常见场景 CREATE TABLE demo_db.orders ( order_id BIGINT, user_id BIGINT, amount DECIMAL(12,2), status VARCHAR(32), create_time DATETIME ) DUPLICATE KEY(order_id) DISTRIBUTED BY HASH(order_id) BUCKETS 8 PROPERTIES (replication_num 3);建表时replication_num在单机测试环境要改成 1否则会因为副本数不足建表失败。生产环境按集群规模设 3。3.4 一个最小可运行的转换示例下面是一个完整的转换逻辑从 CSV 文件读数据经过字段类型转换写入 Doris。# 这不是实际代码是转换的步骤描述用伪代码说明数据流 # 步骤1CSV输入读取 orders.csv分隔符逗号首行为列名 # 步骤2字段选择把 order_id 从 String 转成 Integer # 步骤3Doris Stream Loader配置如下 # feHost fe-host # fePort 8030 # database demo_db # table orders # batchSize 5000 # batchByteSize 5242880 # format csv # columns order_id,user_id,amount,status,create_time实际运行时Kettle 会启动 CSV 读取线程和 Stream Load 写入线程中间通过 RowSet 缓冲。如果 CSV 有 100 万行batchSize 设 5000会分成 200 个批次发送。每个批次是一个独立的 HTTP 请求失败只重试当前批次。3.5 运行与验证保存转换后点运行。Spoon 底部的日志面板会输出每个批次的导入结果包括写入行数、耗时、是否成功。如果看到Stream load success且行数对得上说明链路通了。验证数据是否真的进去了用 MySQL 客户端连 Doris 查一下SELECT COUNT(*) FROM demo_db.orders; -- 对比源文件行数确认没有丢数据如果行数对不上先看日志里有没有filtered rows的提示。Stream Load 会把类型转换失败的行过滤掉但不算整批失败。这个行为可以通过maxFilterRatio参数控制默认 0 表示不允许任何过滤有过滤就整批失败。生产环境建议设成 0.1 到 0.3容忍少量脏数据。4. 参数调优batchSize、并发度和内存的三角关系4.1 batchSize 与 batchByteSize 怎么配这两个参数是乘数关系谁先到阈值谁触发发送。batchSize 管行数batchByteSize 管字节数。如果只设 batchSize 不设 batchByteSize遇到超宽行时单批内存会爆只设 batchByteSize 不设 batchSize遇到超窄行时请求次数会过多。经验配比是行宽 200 字节以内batchSize 设 10000 到 20000batchByteSize 设 10MB 到 20MB行宽 1KB 左右batchSize 降到 3000 到 5000batchByteSize 保持 10MB行宽超过 5KBbatchSize 设 1000 以内batchByteSize 设 20MB 以上让字节数主导。调参时用 Doris 的SHOW LOAD命令看每个批次的实际大小和耗时。如果发现批次普遍只有几百 KB说明 batchSize 设小了如果单批耗时超过 10 秒说明批次太大BE 处理不过来。4.2 并发度怎么开Kettle 的并发靠步骤副本实现。在转换里右键 Stream Loader 步骤选改变开始复制的数量设成 2 到 4。每个副本是一个独立线程各自维护缓冲区向不同的 BE 发请求。但并发不是越高越好。Doris 的单个 Tablet 写入是有锁的如果多个批次命中同一个 Tablet会串行化。并发度超过 BE 数量时多出来的线程只是在排队。一般建议并发度等于 BE 节点数或者 BE 节点数的一半。另一个限制是 Kettle 自身的内存。每个副本的缓冲区大小是 batchByteSize 乘以 2发送中 攒批中4 个副本 10MB 批次就是 80MB 常驻内存。Kettle 默认 JVM 堆是 1G开太多副本会 OOM。4.3 超时与重试参数Stream Load 的 HTTP 请求有超时限制Doris 侧默认 600 秒插件侧默认 300 秒。如果单批数据量大、BE 负载高300 秒可能不够。可以在插件配置里把timeout调到 600和 Doris 侧对齐。重试参数maxRetries默认 3意思是失败后重试 3 次。但要注意Stream Load 的重试不是幂等的——如果第一次请求实际写入了但响应超时重试会导致数据重复。Doris 的 Stream Load 支持通过label参数做幂等插件会自动生成唯一 label重试时用同一个 labelDoris 侧会识别并去重。这个机制默认开启不用额外配置。4.4 内存与 GC 的观察方法调参期间用jstat -gc观察 Kettle 进程的 GC 频率。如果 Full GC 每分钟超过一次说明内存压力大要么降 batchByteSize要么减并发度。用jmap -histo看堆里占比最高的对象正常情况下应该是 byte 数组缓冲区和 Row 对象。有个隐蔽的坑Kettle 的 RowSet 默认大小是 10000 行如果上游生产速度远快于 Stream Load 消费速度RowSet 会满上游步骤会阻塞。这是正常的背压机制但如果阻塞时间过长上游数据库连接可能超时。解决办法是适当增大 RowSet 大小或者在上游加一个阻塞数据步骤做缓冲。5. 避坑指南五条血泪经验5.1 现象导入报tablet writer write failed重试多次仍失败原因通常是 BE 磁盘写满或者 Tablet 副本状态异常。Doris 的 Stream Load 在写入前会检查 Tablet 的健康状态如果有副本处于bad状态写入会被拒。解决分两步。先查SHOW TABLET FROM demo_db.orders看有没有异常副本有的话用ADMIN REPAIR TABLE触发修复。再查 BE 的磁盘使用率超过 85% 就要清理或扩容。临时规避可以把replication_num降到 1但生产环境不建议。5.2 现象数据行数对得上但部分字段是 NULL原因是 CSV 的列顺序和 Doris 表的列顺序不一致或者 JSON 的 key 名和列名大小写不匹配。Stream Load 在列映射失败时不会报错而是把对应列填 NULL。解决方法是显式指定columns参数把上游字段名和 Doris 列名的映射写清楚。CSV 模式下columns的顺序就是数据里字段的顺序不是表的顺序。JSON 模式下确保 key 名和列名完全一致包括大小写。5.3 现象导入速度一开始很快跑几分钟后骤降原因是 Doris 在后台做 Compaction。Stream Load 写入的是增量数据Doris 会定期把多个小文件合并成大文件。Compaction 期间磁盘 IO 和 CPU 被占用导入速度会下降。这是正常现象不用干预。如果下降幅度超过 50% 且持续超过 10 分钟说明 Compaction 压力过大可能是批次太小导致小文件太多。把 batchSize 调大减少批次数量能从源头缓解。5.4 现象Kettle 报OutOfMemoryError: Java heap space原因是并发度乘以 batchByteSize 超过了 JVM 堆。比如 4 个副本、每个 20MB 批次加上 RowSet 缓冲和 Kettle 自身开销1G 堆不够用。解决办法有两个一是改 Kettle 启动脚本里的-Xmx调到 2G 或 4G二是降并发度或 batchByteSize。优先调 JVM 参数因为降参数会影响吞吐。改spoon.sh里的PENTAHO_DI_JAVA_OPTIONS加上-Xmx4g。5.5 现象任务卡在 Stream Loader 步骤不动日志无输出原因是 FE 地址配错或者 FE 到 BE 的重定向失败。插件在启动时会向 FE 请求 BE 列表如果 FE 不可达或者返回的 BE 地址客户端连不上就会一直卡在初始化阶段。排查方法先用 curl 手动请求 FE 的/api/backends接口确认能返回 BE 列表。再用 curl 直接向 BE 发一个测试 Stream Load 请求确认网络通。如果 FE 返回的 BE 地址是内网 IP 而客户端在外网需要在插件里配置beAddress参数做地址替换。6. 进阶用 label 做幂等重跑与断点续传的边界跑通基础链路之后真正决定这套方案能不能上生产的是失败重跑时的数据一致性。Stream Load 的 label 机制是这里的关键。每个导入请求带一个 labelDoris 会记录 label 和导入结果的对应关系。同一个 label 重复提交Doris 不会重复写入而是返回第一次的结果。插件默认用kettle_{步骤名}_{时间戳}_{随机数}生成 label这个 label 在单次运行内唯一但任务重跑时会变所以重跑会重复写入。要支持幂等重跑得让 label 可预测。常见做法是用业务字段拼 label比如orders_{日期}_{批次号}。插件支持在配置里指定 label 前缀配合 Kettle 的变量替换可以做到同一天同一批次重跑时 label 不变。# 手动验证 label 幂等性 # 第一次提交label 为 test_label_001 curl -X PUT http://be-host:8040/api/demo_db/orders/_stream_load \ -H label: test_label_001 \ -H format: csv \ -T data.csv # 第二次提交同一个 labelDoris 返回上次结果不重复写入 curl -X PUT http://be-host:8040/api/demo_db/orders/_stream_load \ -H label: test_label_001 \ -H format: csv \ -T data.csv但 label 幂等有个边界它只保证同一个 label 不重复写入不保证数据是最新的。如果第一次导入的数据有误用同一个 label 重跑不会覆盖而是直接返回成功。要修正数据必须换一个新 label然后手动删除旧数据。这个边界在设计和重跑策略时必须想清楚。另一个进阶用法是配合 Doris 的 Routine Load 做兜底。Stream Load 适合批量同步Routine Load 适合持续消费 Kafka。如果业务既有批量补数又有实时增量可以用 Stream Load 做补数、Routine Load 做增量两者写同一张表Doris 会自动处理版本合并。最后说一个我自己的习惯每次调完参数不要只看单次运行的耗时要连续跑三次取平均值。Doris 的 Compaction 和 Kettle 的 JIT 预热都会影响第一次运行的结果单次数据很容易误导判断。我一般会在转换里加一个统计步骤把每批的耗时和行数写到日志表里跑一周后看趋势比单次调参靠谱得多。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
