Flink实时写、离线批量导、维度表天天更新——Doris数据导入方式全解析

Doris手册第 4 题约 10 分钟读完

本文目录

先说结论
四种导入方式对比
Stream Load——实时写入的首选
Broker Load——大批量离线导入
Insert Into——简单但慎用
Routine Load——Doris原生消费Kafka
实战踩坑：Insert Into导致的小文件风暴
面试高频追问
总结

先说结论

Doris有四种主要的数据导入方式：Stream Load（实时/微批，最常用）、Broker Load（大批量离线导入）、Insert Into（SQL插入，小数据量）、Routine Load（自动消费Kafka）。选型核心原则：实时场景用Stream Load、离线批量用Broker Load、临时小量用Insert Into。

先把这段讲顺；面试官追问时，再展开下面的原理和细节。

详细讲解

Doris的数据导入方式有好几种，每种适用场景不同。在某元宇宙社交公司做实时数仓时，Flink实时写Doris用的是Stream Load，离线批量导入历史数据用的是Broker Load，小量维度表更新用Insert Into。面试官问这道题，就是想看你知不知道什么场景用什么方式，以及有没有踩过导入相关的坑。

四种导入方式对比

方式	原理	适用场景	吞吐量	实时性
Stream Load	HTTP API推送数据到BE	Flink/Spark实时写入	高（万条/秒）	秒级
Broker Load	BE从HDFS/S3拉取数据	离线批量导入历史数据	极高	分钟-小时
Insert Into	SQL INSERT语句	小数据量/临时导入	低	秒级
Routine Load	Doris自动消费Kafka	Kafka实时入库	中等	秒级

Stream Load——实时写入的首选

Stream Load是我用得最多的导入方式。原理很简单：客户端通过HTTP PUT请求把数据推送给Doris的BE节点，BE解析数据后直接写入。

# Stream Load示例（命令行方式）
curl -u root:password \
    -H 'format: json' \
    -H 'strip_outer_array: true' \
    -H 'jsonpaths: ["$.event_time","$.user_id","$.event_name"]' \
    -T data.json \
    http://fe-host:8030/api/mydb/ods_user_event/_stream_load
 
# Flink写Doris的方式（生产中最常用）
# ① Flink Doris Connector（官方推荐）
#    底层封装了Stream Load，支持Exactly-Once（2PC）
# ② 手动调用Stream Load API
#    在Flink的ProcessFunction里攒批，然后HTTP调用
#    更灵活但需要自己处理重试和幂等
 
# 性能参考（我在元宇宙社交公司的实际数据）：
# 单BE节点Stream Load吞吐：约5000-10000条/秒（JSON格式）
# 攒批1000条或5秒一批，延迟在5-10秒级别

Stream Load的关键配置：

// Flink Doris Connector配置（官方推荐方式）
DorisSink.sink(
    DorisExecutionOptions.builder()
        .setStreamLoadProp(new Properties() {{
            put("format", "json");
            put("strip_outer_array", "true");
        }})
        .setBatchSize(1000)       // 攒够1000条触发写入
        .setBatchIntervalMs(5000) // 或5秒触发写入
        .setMaxRetries(3)         // 失败重试3次
        .build(),
    DorisOptions.builder()
        .setFenodes("fe-host:8030")
        .setTableIdentifier("mydb.ods_user_event")
        .setUsername("root")
        .setPassword("password")
        .build()
);

Broker Load——大批量离线导入

当你需要把HDFS/S3上的大量历史数据导入Doris时，用Broker Load。它的原理是BE节点主动从外部存储拉取数据，适合一次性导入几十GB到几TB的数据。

-- Broker Load：从S3导入历史数据
LOAD LABEL mydb.load_history_20240115 (
    DATA INFILE('s3://my-bucket/data/user_events/dt=2024-01-15/*')
    INTO TABLE ods_user_event
    COLUMNS TERMINATED BY ','
    (event_time, event_id, user_id, event_name, properties)
)
WITH S3 (
    'AWS_ENDPOINT' = 's3.us-west-2.amazonaws.com',
    'AWS_ACCESS_KEY' = 'xxx',
    'AWS_SECRET_KEY' = 'xxx'
)
PROPERTIES (
    'timeout' = '3600',        -- 1小时超时
    'max_filter_ratio' = '0.1' -- 允许10%错误率
);
 
-- 查看导入状态
SHOW LOAD WHERE LABEL = 'load_history_20240115';

Insert Into——简单但慎用

Insert Into适合小数据量的场景，比如手动插几条测试数据、从一张Doris表查询结果写入另一张表。

-- 小数据量插入
INSERT INTO dim_gift_config VALUES
    (1, 'rose', 9.9, 'flower'),
    (2, 'bear', 19.9, 'toy'),
    (3, 'rocket', 99.9, 'premium');
 
-- 表到表（ETL场景，适合DWS层建设）
INSERT INTO dws_daily_overview
SELECT dt, COUNT(DISTINCT user_id) AS uv, COUNT(*) AS pv,
       SUM(amount) AS total_amount
FROM ods_user_event
WHERE dt = '2024-01-15'
GROUP BY dt;
 
-- ⚠️ 注意：不要用INSERT INTO做大批量实时写入！
-- 每次INSERT都会生成一个小的数据版本
-- 高频INSERT → 小文件爆炸 → Compaction压力大 → 查询变慢

Routine Load——Doris原生消费Kafka

Routine Load让Doris直接消费Kafka的数据，不需要Flink等中间件。适合简单场景：

-- 创建Routine Load任务
CREATE ROUTINE LOAD mydb.kafka_load ON ods_user_event
COLUMNS(event_time, event_id, user_id, event_name, properties)
PROPERTIES (
    'format' = 'json',
    'jsonpaths' = '["$.event_time","$.event_id","$.user_id","$.event_name","$.properties"]',
    'desired_concurrent_number' = '3',  -- 3个消费线程
    'max_batch_interval' = '10',        -- 10秒一批
    'max_batch_rows' = '100000'         -- 或10万条一批
)
FROM KAFKA (
    'kafka_broker_list' = 'broker1:9092,broker2:9092',
    'kafka_topic' = 'user_event',
    'property.group.id' = 'doris-routine-load'
);
 
-- 查看任务状态
SHOW ROUTINE LOAD FOR mydb.kafka_load;

Routine Load vs Flink+Stream Load：Routine Load更简单（不需要Flink），但不支持复杂ETL（过滤、Join、聚合）。如果需要在入Doris前做数据清洗和转换，还是要用Flink+Stream Load。我在元宇宙社交公司选的是Flink+Stream Load，因为入库前需要做维表Join和脏数据过滤。

实战踩坑：Insert Into导致的小文件风暴

早期在某元宇宙社交公司，有个同事写了一个Python脚本，每5秒用Insert Into往Doris写一批数据（模拟实时写入）。跑了两天后Doris的BE节点磁盘IO飙升，查询速度从毫秒级降到秒级。

排查发现：每次Insert Into都生成一个独立的数据版本（一组小文件）。两天下来产生了几万个小版本，BE的Compaction（后台合并）线程跟不上，小文件太多导致查询要打开大量文件，IO暴增。

解决方案：把Insert Into改成Stream Load，攒批1000条或5秒一批。Stream Load单次写入的数据会合并成一个版本，大幅减少版本数量。改完后BE的Compaction压力立刻降下来，查询恢复到毫秒级。

教训：Doris的实时写入必须用Stream Load，绝对不要用高频Insert Into。Insert Into只适合临时的小数据量操作。

面试高频追问

追问一：Stream Load支持Exactly-Once吗？

支持。Doris的Stream Load支持两阶段提交（2PC）：先预提交（数据写入但不可见），Flink Checkpoint成功后再正式提交。如果Checkpoint失败则回滚。Flink Doris Connector默认启用2PC。

追问二：Doris导入数据时怎么做数据质量校验？

Stream Load和Broker Load都支持max_filter_ratio参数，允许一定比例的错误行被跳过。导入完成后返回的结果里包含error_url，可以查看被过滤掉的具体行和原因。生产环境建议设0.01（1%错误率），超过就告警。

追问三：Routine Load和Flink+Stream Load怎么选？

简单场景（Kafka直接入Doris，不需要ETL）用Routine Load，运维更简单。复杂场景（入库前要清洗、Join维表、过滤脏数据）用Flink+Stream Load。我的经验是生产环境大部分场景需要ETL，所以Flink+Stream Load用得更多。

总结

Doris导入方式面试记住四种：实时用Stream Load、大批量用Broker Load、临时小量用Insert Into、简单Kafka入库用Routine Load。核心踩坑点：不要用高频Insert Into做实时写入（小文件风暴），用Stream Load攒批代替。能讲出Flink+Stream Load的配合方式和2PC的Exactly-Once保证，面试就很有深度了。

下一篇聊一个经典对比题：Doris vs ClickHouse vs StarRocks，为什么很多公司选Doris而不是ClickHouse？

你们Doris的数据导入用什么方式？有没有因为导入方式踩过坑？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长

刷题进度保存在当前浏览器