未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

warning 适用范围与待维护事项

本文用于已有Doris、Kafka及ETL经验者比较导入方案，不是可直接执行的生产脚本。老王案例数字为原稿未核实的教学假设；2GB/5GB、秒级延迟及并发建议不是固定限制。未指定Doris和DataX插件版本，也未提供表DDL与样本；列转换、事务、Broker及系统参数须逐项验证。Python重定向认证和错误处理尚不完整；root:root与明文HTTP仅为占位示例，允许过滤错误行时必须核查拒绝记录并对账。

Doris数据导入方案详解 - 多渠道数据接入指南

abstract 📋 本节概览

🎯 学习目标：掌握Doris各种数据导入方式的原理与应用
📚 前置知识：08-表结构设计与优化|表结构设计与优化
⏱️ 预计用时：30分钟

🚛 数据搬运工的"噩梦"

以下为教学假设故事，并非已核实的企业案例；200GB、10GB、8小时、1小时、2GB和30分钟均为原稿未核实的示例数字，不代表性能基准或产品限制。假设电商公司数据工程师老王接到新任务：把每天200GB的业务数据导入Doris分析。

第一天：用最简单的INSERT语句

INSERT INTO user_behavior SELECT * FROM mysql.raw_data;

结果：导入10GB用了8小时，系统差点挂了。

第二天：改用批量导入

# Stream Load方式
curl -X PUT "http://doris:8030/api/database/table/_stream_load" \
-H "column_separator:," -T data.csv

假设结果：1小时完成，该示例环境将单次导入量设为2GB；这不是Doris的通用上限。

第三天：发现Broker Load

LOAD LABEL example_db.label1 (
    DATA INFILE("hdfs://path/to/bigdata/*") 
    INTO TABLE user_behavior
);

结果：200GB数据，30分钟搞定！

老王感慨（教学假设）："导入方式应按场景选择，性能差异需要在相同数据、资源和配置下实测，不能由上述示例推导出100倍的结论。"

这就是为什么要掌握多种数据导入方案——不同场景需要不同策略。

💡 Doris数据导入方式概览

导入方式	适用场景	数据量级	实时性	可靠性
Stream Load	小批量实时导入	MB-GB	秒级	高
Broker Load	大批量离线导入	GB-TB	分钟级	高
Routine Load	流式持续导入	不限	秒级	高
Insert Into	小批量数据写入	MB级	秒级	高
S3 Load	云存储大批量导入	TB级	分钟级	高
DataX/DTS	异构数据源同步	不限	分钟级	高

🚀 Stream Load - 小批量实时导入

原理与特点

note 核心原理

Stream Load通过HTTP协议将本地文件或数据流直接发送到Doris FE/BE节点，适合小批量数据的快速导入。

主要特点：

同步执行，等待导入结果
适合小批量数据（通常<5GB）
支持CSV、JSON格式
支持HTTP压缩传输

实战示例

# 基本语法
curl --location-trusted -u username:password \
    -H "label:label_name" \
    -H "column_separator:," \
    -H "columns: col1, col2, col3" \
    -T data_file.csv \
    http://fe_host:http_port/api/db_name/table_name/_stream_load

# 实际示例：导入CSV文件
curl --location-trusted -u root:root \
    -H "label:user_data_20230601" \
    -H "column_separator:," \
    -H "columns: user_id, user_name, age, register_time" \
    -T user_data.csv \
    http://doris-fe:8030/api/user_db/user_table/_stream_load

Python SDK调用示例：

import requests
import json
import time

def stream_load(file_path, host, port, db, table, user, passwd):
    """
    使用Stream Load导入数据到Doris
    """
    url = f"http://{host}:{port}/api/{db}/{table}/_stream_load"
    
 # 设置请求头
    headers = {
        'label': f"python_streamload_{db}_{table}_{int(time.time())}",
        'column_separator': ',',
        'columns': 'user_id, user_name, age, register_time'
    }
    
 # 读取文件内容
    with open(file_path, 'rb') as f:
        data = f.read()
    
 # 发送请求
    response = requests.put(
        url,
        auth=(user, passwd),
        headers=headers,
        data=data
    )
    
 # 解析结果
    result = json.loads(response.text)
    if result.get('Status') == 'Success':
        print(f"导入成功，导入数据量: {result.get('NumberLoadedRows')} 行")
    else:
        print(f"导入失败: {result.get('Message')}")
    
    return result

# 调用示例
stream_load(
    file_path="user_data.csv",
    host="doris-fe",
    port=8030,
    db="user_db",
    table="user_table",
    user="root",
    passwd="root"
)

高级配置

# 指定JSON格式
curl --location-trusted -u root:root \
    -H "label:json_data_20230601" \
    -H "format:json" \
    -H "jsonpaths:[\"$.user_id\", \"$.name\", \"$.age\"]" \
    -H "strip_outer_array:true" \
    -T user_data.json \
    http://doris-fe:8030/api/user_db/user_table/_stream_load

# 部分列导入与转换
curl --location-trusted -u root:root \
    -H "label:partial_data_20230601" \
    -H "column_separator:," \
    -H "columns: user_id, user_name, age, register_time, temp" \
    -H "where: age >= 18" \
    -H "function_column.city_code: city_name_to_code(user_city)" \
    -T user_data.csv \
    http://doris-fe:8030/api/user_db/user_table/_stream_load

🚚 Broker Load - 大批量离线导入

原理与特点

note 核心原理

Broker Load通过部署的Broker进程访问外部存储系统（HDFS、S3等），读取数据并导入Doris，适合大批量离线数据导入。

主要特点：

异步执行，支持后台导入
适合大批量数据（GB-TB级）
支持HDFS、S3、阿里云OSS等外部存储
支持多种文件格式（CSV、ORC、Parquet）

实战示例

-- 创建导入作业
LOAD LABEL user_db.broker_load_20230601
(
    DATA INFILE("hdfs://hadoop-cluster/user/doris/data/user_*.csv")
    INTO TABLE user_table
    COLUMNS TERMINATED BY ","
    (user_id, user_name, age, register_time)
)
WITH BROKER
(
    "username" = "hdfs_user",
    "password" = "hdfs_password"
)
PROPERTIES
(
    "timeout" = "3600",
    "max_filter_ratio" = "0.1"
);

-- 查看导入状态
SHOW LOAD FROM user_db WHERE LABEL = "broker_load_20230601";

导入Parquet/ORC格式

-- 导入Parquet文件
LOAD LABEL user_db.parquet_load_20230601
(
    DATA INFILE("hdfs://hadoop-cluster/user/doris/data/user_data.parquet")
    INTO TABLE user_table
    FORMAT AS "parquet"
)
WITH BROKER
(
    "username" = "hdfs_user",
    "password" = "hdfs_password"
)
PROPERTIES
(
    "timeout" = "3600",
    "max_filter_ratio" = "0.1"
);

-- 导入ORC文件
LOAD LABEL user_db.orc_load_20230601
(
    DATA INFILE("hdfs://hadoop-cluster/user/doris/data/user_data.orc")
    INTO TABLE user_table
    FORMAT AS "orc"
)
WITH BROKER
(
    "username" = "hdfs_user",
    "password" = "hdfs_password"
);

🔄 Routine Load - 流式持续导入

原理与特点

note 核心原理

Routine Load建立与Kafka等消息队列的长连接，持续消费数据并导入Doris，适合实时数据流的持续导入。

主要特点：

持续执行，自动消费数据
支持Kafka数据源
支持断点续传和数据一致性
支持多种格式（CSV、JSON）

实战示例

-- 创建Kafka持续导入任务
CREATE ROUTINE LOAD user_db.kafka_load_orders ON orders_table
COLUMNS(order_id, user_id, product_id, order_time, amount)
PROPERTIES
(
    "format" = "csv",
    "column_separator" = ",",
    "jsonpaths" = "",
    "max_batch_interval" = "10",
    "max_batch_rows" = "200000",
    "max_batch_size" = "100000000"
)
FROM KAFKA
(
    "kafka_broker_list" = "kafka1:9092,kafka2:9092,kafka3:9092",
    "kafka_topic" = "orders_topic",
    "kafka_partitions" = "0,1,2,3,4",
    "kafka_offsets" = "OFFSET_BEGINNING,OFFSET_BEGINNING,OFFSET_BEGINNING,OFFSET_BEGINNING,OFFSET_BEGINNING",
    "property.group.id" = "doris_orders_group",
    "property.client.id" = "doris_orders_client"
);

-- 查看导入状态
SHOW ROUTINE LOAD;

-- 暂停导入任务
PAUSE ROUTINE LOAD FOR user_db.kafka_load_orders;

-- 恢复导入任务
RESUME ROUTINE LOAD FOR user_db.kafka_load_orders;

-- 停止导入任务
STOP ROUTINE LOAD FOR user_db.kafka_load_orders;

JSON格式导入配置

-- 创建Kafka JSON格式导入任务
CREATE ROUTINE LOAD user_db.kafka_json_load ON user_behavior_table
COLUMNS(user_id, behavior_type, product_id, timestamp)
PROPERTIES
(
    "format" = "json",
    "jsonpaths" = "[\"$.user_id\", \"$.behavior\", \"$.product_id\", \"$.ts\"]",
    "strip_outer_array" = "true",
    "max_batch_interval" = "10",
    "max_batch_rows" = "200000"
)
FROM KAFKA
(
    "kafka_broker_list" = "kafka1:9092,kafka2:9092",
    "kafka_topic" = "user_behavior",
    "kafka_partitions" = "0,1,2",
    "kafka_offsets" = "OFFSET_END,OFFSET_END,OFFSET_END"
);

📝 Insert Into - 灵活数据写入

原理与特点

note 核心原理

通过SQL的Insert语句将数据写入Doris表，适合小批量数据操作或表间数据转换。

主要特点：

支持单条或批量插入
支持从其他表查询结果插入
适合小批量数据或表间ETL

实战示例

-- 单条数据插入
INSERT INTO user_table VALUES (1001, 'Zhang San', 25, '2023-01-15');

-- 批量数据插入
INSERT INTO user_table VALUES 
    (1002, 'Li Si', 30, '2023-01-16'),
    (1003, 'Wang Wu', 28, '2023-01-17'),
    (1004, 'Zhao Liu', 35, '2023-01-18');

-- 从查询结果插入
INSERT INTO user_summary_table
SELECT 
    DATE_FORMAT(register_time, '%Y-%m-%d') as day,
    COUNT(*) as new_users,
    AVG(age) as avg_age
FROM user_table
GROUP BY DATE_FORMAT(register_time, '%Y-%m-%d');

高级应用

-- 使用PREPARE语句进行参数化插入
PREPARE stmt FROM "INSERT INTO user_table VALUES (?, ?, ?, ?)";
SET @user_id = 1005;
SET @user_name = 'Zhou Qi';
SET @age = 40;
SET @register_time = '2023-01-19';
EXECUTE stmt USING @user_id, @user_name, @age, @register_time;

-- 事务插入
BEGIN TRANSACTION;
INSERT INTO orders_table VALUES (10001, 1001, 2001, '2023-06-01 10:00:00', 99.99);
INSERT INTO order_items_table VALUES (10001, 1, 2001, 1, 99.99);
COMMIT;

🔄 S3 Load - 云存储批量导入

原理与特点

note 核心原理

直接从S3兼容的云存储读取数据并导入Doris，无需额外的Broker组件。

主要特点：

支持Amazon S3、阿里云OSS、腾讯云COS等
支持多种文件格式（CSV、Parquet、ORC）
性能需结合存储服务、网络、文件布局及Doris版本实测，不能笼统认定优于Broker Load

实战示例

-- 从S3导入数据
LOAD LABEL user_db.s3_load_20230601
(
    DATA INFILE("s3://doris-bucket/data/user_*.csv")
    INTO TABLE user_table
    COLUMNS TERMINATED BY ","
    (user_id, user_name, age, register_time)
)
WITH S3
(
    "s3.endpoint" = "https://s3.amazonaws.com",
    "s3.access_key" = "your_access_key",
    "s3.secret_key" = "your_secret_key",
    "s3.region" = "us-west-1"
)
PROPERTIES
(
    "timeout" = "3600",
    "max_filter_ratio" = "0.1"
);

🔄 DataX/DTS - 异构数据源集成

原理与特点

note 核心原理

利用DataX、DTS等ETL工具从各种数据源抽取数据并导入Doris。

主要特点：

支持多种数据源（MySQL、Oracle、PostgreSQL等）
支持全量和增量同步
可实现复杂的数据转换

DataX配置示例

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "username": "mysql_user",
                        "password": "mysql_password",
                        "column": ["id", "name", "age", "register_time"],
                        "connection": [
                            {
                                "table": ["user_table"],
                                "jdbcUrl": ["jdbc:mysql://mysql-host:3306/source_db"]
                            }
                        ]
                    }
                },
                "writer": {
                    "name": "doriswriter",
                    "parameter": {
                        "feLoadUrl": ["doris-fe:8030"],
                        "beLoadUrl": ["doris-be1:8040", "doris-be2:8040"],
                        "jdbcUrl": "jdbc:mysql://doris-fe:9030/user_db",
                        "database": "user_db",
                        "table": "user_table",
                        "column": ["user_id", "user_name", "age", "register_time"],
                        "username": "root",
                        "password": "root",
                        "postSql": [],
                        "preSql": [],
                        "loadProps": {
                            "format": "csv",
                            "column_separator": ",",
                            "row_delimiter": "\n"
                        },
                        "maxBatchRows": 200000,
                        "maxBatchSize": 104857600
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": 4
            }
        }
    }
}

🔍 导入性能优化

1. 批量导入策略

批次大小优化：可将50MB-1GB作为原稿给出的试验区间，而非通用最优值；应按导入方式、内存和超时配置压测
并行度控制：可将BE节点数 * 2作为原稿给出的并发试验起点，而非固定公式；应监测CPU、内存、磁盘和查询延迟后调整
合并小文件：避免大量小文件导入，提前合并

2. 表设计优化

-- 优化分区设计，减少数据导入时的写放大
CREATE TABLE optimized_table (
    event_time DATETIME,
    user_id BIGINT,
    event_type VARCHAR(20),
    data VARCHAR(1024)
)
PARTITION BY RANGE(event_time) (
    PARTITION p20230601 VALUES [('2023-06-01'), ('2023-06-02'))
)
DISTRIBUTED BY HASH(user_id) BUCKETS 16;

-- 设置合理的导入参数
LOAD LABEL user_db.optimized_load
(
    DATA INFILE("hdfs://hadoop-cluster/user/doris/data/*.csv")
    INTO TABLE optimized_table
)
WITH BROKER
(
    "username" = "hdfs_user",
    "password" = "hdfs_password"
)
PROPERTIES
(
    "timeout" = "3600",
    "max_filter_ratio" = "0.1",
    "send_batch_parallelism" = "5",
    "load_parallelism" = "8"
);

3. 系统参数调优

-- 调整导入并发度
ADMIN SET FRONTEND CONFIG ("max_running_txn_num_per_db" = "10");

-- 调整BE节点导入并发度
ADMIN SET BACKEND CONFIG ("push_worker_count_normal_priority" = "8");
ADMIN SET BACKEND CONFIG ("push_worker_count_high_priority" = "16");

⚠️ 常见问题与解决方案

1. 导入超时

问题：大批量数据导入时出现超时错误
解决方案：
- 增加导入任务的timeout参数值
- 将数据拆分为多个小批次导入
- 调整BE节点的内存和CPU资源

-- 设置更长的超时时间
LOAD LABEL user_db.large_load
(
    DATA INFILE("hdfs://hadoop-cluster/user/doris/large_data/*.csv")
    INTO TABLE large_table
)
WITH BROKER
(
    "username" = "hdfs_user",
    "password" = "hdfs_password"
)
PROPERTIES
(
    "timeout" = "86400"  -- 设置为24小时
);

2. 数据质量问题

问题：导入数据包含不符合表结构的数据
解决方案：
- 设置合理的max_filter_ratio允许部分错误数据
- 使用where条件过滤不合格数据
- 添加数据转换函数处理异常值

-- 设置过滤比例和数据转换
LOAD LABEL user_db.quality_control_load
(
    DATA INFILE("hdfs://hadoop-cluster/user/doris/data/*.csv")
    INTO TABLE user_table
    COLUMNS TERMINATED BY ","
    (user_id, user_name, age_str, register_time)
)
WITH BROKER
(
    "username" = "hdfs_user",
    "password" = "hdfs_password"
)
PROPERTIES
(
    "max_filter_ratio" = "0.2",
    "where" = "age_str != 'NULL'",
    "function_column.age" = "cast(age_str as INT)"
);

3. Routine Load 消费延迟

问题：Kafka消费速度跟不上生产速度
解决方案：
- 增加任务并行度
- 调整批次参数
- 优化表结构减少写放大

-- 优化Routine Load配置
ALTER ROUTINE LOAD user_db.kafka_load_orders
PROPERTIES
(
    "desired_concurrent_number" = "5",  -- 增加并行度
    "max_batch_interval" = "5",         -- 减少批次间隔
    "max_batch_rows" = "500000",        -- 增加批次大小
    "max_batch_size" = "209715200"      -- 增加到200MB
);

✅ 掌握检查

[ ] 能根据场景选择合适的数据导入方式
[ ] 掌握Stream Load的基本用法
[ ] 了解Broker Load的配置与优化
[ ] 能配置Routine Load实现实时数据同步
[ ] 掌握导入性能优化的关键参数

🔗 学习连接

前置知识: 表结构设计与优化 下一步: SQL查询与