Apache Doris 数据导入实践：四大导入方式与性能调优指南
Apache Doris 是一款高性能的实时分析型数据库，支持多种数据导入方式。本文将详细介绍 Doris 的三种主要数据导入方式：Broker Load、Spark Load 和 Insert Into，并深入探讨导入性能调优的关键技巧。
1. Doris 数据导入方式概述
Apache Doris 提供了多种数据导入方式，以满足不同场景下的数据加载需求。了解各种导入方式的特点和适用场景，是选择合适导入策略的第一步。
| 导入方式 | 特点 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|---|
| Broker Load | 基于 Broker 进程的分布式导入，异步执行 | 大批量数据导入，TB 级数据 | 高并发、高吞吐、可暂停恢复 | 需要部署 Broker 服务，导入延迟较高 |
| Spark Load | 基于 Spark 计算引擎的导入，支持复杂转换 | 需要数据预处理或复杂转换场景 | 支持复杂数据转换，可扩展性强 | 配置复杂，依赖 Spark 环境 |
| Insert Into | 标准 SQL 插入，同步执行 | 小批量实时数据插入 | 实时性好，使用简单 | 不适合大批量数据，性能较低 |
流程图展示了 Doris 数据导入的基本流程：
2. Broker Load 实践与优化
Broker Load 是 Doris 提供的一种高效的大批量数据导入方式，基于 Broker 进程实现数据的并行加载。
2.1 基本语法与使用
Broker Load 使用以下基本语法：
LOAD LABEL database.table_label
(
DATA INFILE("hdfs://path/to/data/*")
INTO TABLE table_name
[PARTITION(p1)]
[COLUMNS(col1, col2, col3)]
[SET(col4 = col1 * 10)]
)
PROPERTIES
(
"broker.name" = "broker_name",
"broker.hadoop.default.fs" = "hdfs://namenode:8020",
"timeout_second" = "3600"
);2.2 关键参数优化
Broker Load 的性能受多个参数影响，主要包括：
- 并发度：通过
parallel_fragment_exec_instance_num参数控制，通常设置为 CPU 核心数的 2-3 倍 - 内存限制：通过
exec_mem_limit参数设置，建议为总内存的 20%-30% - 超时时间：根据数据量大小合理设置，避免任务因超时失败
2.3 常见问题与解决方案
- 导入失败：检查 Broker 服务是否正常，文件路径是否存在且可访问
- 导入速度慢：提高并发度，优化数据格式（如使用列式存储 Parquet）
- 内存不足：适当调小
exec_mem_limit或增加 BE 节点
3. Spark Load 实践与优化
Spark Load 利用 Spark 分布式计算能力，适合处理需要复杂数据转换的大规模数据导入场景。
3.1 基本语法与配置
Spark Load 的基本语法如下：
LOAD LABEL database.table_label
(
SPARK
"spark.sql"
(
"spark.executor.memory"="4g",
"spark.executor.cores"="2",
"spark.executor.instances"="10"
)
SCRIPT
"spark_load.py"
-- "hdfs://path/to/data/*"
"hdfs://output/path"
)
INTO TABLE table_name
[PARTITION(p1)]
[COLUMNS(col1, col2, col3)]
[SET(col4 = expr)];3.2 Spark 优化要点
优化 Spark Load 性能的关键点：
- 资源配置：根据数据量调整 executor 数量和内存，通常 executor 数量设置为集群核心数的 1-2 倍
- 数据倾斜处理：对关键字段进行预分发，避免数据倾斜
- 序列化优化：使用 Kryo 序列化替代默认 Java 序列化
3.3 典型 Spark 导入脚本示例
# spark_load.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("SparkLoadExample").getOrCreate()
# 读取数据
df = spark.read.parquet("hdfs://input/path/*")
# 数据转换
df_transformed = df.withColumn("new_col", col("old_col") * 2)
# 写入 Doris
df_transformed.write.format("doris").option("doris.table.identifier", "database.table_name").option("doris.fenodes", "fe_host:9030").save()
spark.stop()4. Insert Into 与综合性能调优
4.1 Insert Into 使用方法
Insert Into 是最直接的 SQL 数据插入方式，适合小批量实时数据插入：
-- 单行插入
INSERT INTO table_name VALUES (1, "name", 100);
-- 批量插入
INSERT INTO table_name VALUES
(1, "name1", 100),
(2, "name2", 200),
(3, "name3", 300);
-- 从查询结果插入
INSERT INTO table_name SELECT * FROM source_table WHERE condition;4.2 综合性能调优策略
针对不同导入场景的综合优化策略：
| 场景 | 推荐导入方式 | 关键优化点 |
|---|---|---|
| TB 级批量数据 | Broker Load | 高并发度、列式存储、合理分区 |
| 复杂数据转换 | Spark Load | 资源优化、倾斜处理、并行度调整 |
| 实时小批量数据 | Insert Into | 批量提交、索引优化 |
| 混合场景组合使用 | 组合使用 | 合理任务调度、错峰导入 |
4.3 通用性能优化技巧
- 表结构优化：合理设计列顺序，高频查询前置列
- 分区策略：根据查询模式选择合适的分区键
- 数据编码：对基数低的列使用字典编码
- 批处理大小：Insert Into 时合理控制批次大小（1000-5000 行/批）
- 资源监控：实时监控导入资源使用情况，及时调整参数
最小示例与注意事项
Broker Load 最小示例
-- 1. 在 HDFS 上准备好数据文件
-- 2. 创建目标表
CREATE TABLE sales_data (
date DATE,
product_id INT,
amount DECIMAL(10,2)
) DISTRIBUTED BY HASH(product_id) BUCKETS 10;
-- 3. 执行 Broker Load
LOAD LABEL example.sales_label
(
DATA INFILE("hdfs://data/sales/*")
INTO TABLE sales_data
COLUMNS TERMINATED BY ","
)
PROPERTIES
(
"broker.name" = "broker_hdfs",
"broker.hadoop.default.fs" = "hdfs://namenode:8020",
"timeout_second" = "3000"
);
-- 4. 查看导入状态
SHOW LOAD WHERE LABEL = "sales_label";注意事项
- Broker Load 注意事项：
- 确保 Broker 服务正常运行
- 大文件提前分割为小文件（建议 100M-1G）
- 避免导入期间修改表结构
- Spark Load 注意事项：
- 确保 Spark 和 Doris 版本兼容
- 导入前进行小规模测试验证
- 合理设置 Spark 内存和并行度
- Insert Into 注意事项：
- 避免单行提交，使用批量插入
- 高峰期限制提交频率
- 及时处理导入失败的情况
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/qq_41840843/article/details/166791813
