【免费下载链接】doris
Apache Doris is an easy-to-use, high performance and unified analytics database.
项目地址：
https://gitcode.com/gh_mirrors/dori/doris
点击查看 免费下载
本文面向希望快速上手 Apache Doris 数据接入与扩展能力的开发者，系统解读官方代码示例集 samples/doris-demo/README.md 中的全部实战场景：包括基于 Flink Doris Connector 的实时入仓（MySQL CDC / Kafka / HDFS / Elasticsearch）、基于 Spark Doris Connector 的读取、基于 Java HttpClient 的 Stream Load 导入、基于 Spring + MyBatis 的 JDBC 数据 API，以及 Java / C++ / Python 三种语言的 RPC UDF 函数服务。读完本文，你将获得每个示例从建表、配置到编译运行的全套可直接复用的方案，并理解其底层源码实现原理。
一、doris-demo 概览：官方示例代码集的结构与定位
doris-demo 是 Apache Doris 官方提供的示例代码集合，其目标正如仓库 README 所述：帮助开发者快速使用和掌握 Doris，提供大量 Doris 相关场景的代码示例（原文：“this project provides a large number of code examples of Doris related scenarios”）。
整个目录位于 samples/doris-demo/，各子项目与 README 条目一一对应，结构如下：
| flink-demo | Java | Flink 与 Doris 集成，Flink Doris Connector 读写 Doris | flink-demo/docs/instructions.md |
| spark-demo | Java / Scala | Spark 与 Doris 集成，Spark Doris Connector 读取 Doris | spark-demo/docs/instructions.md |
| stream-load-demo | Java | Stream Load 导入，支持文件（CSV）与内存数据（JSON），基于 Java HttpClient | stream-load-demo/docs/instructions.md |
| spring-jdbc-demo | Java（SpringBoot + MyBatis） | 通过 JDBC 访问 Doris，对外提供 REST 数据接口 | spring-jdbc-demo/docs/instructions.md |
| remote-udf-java-demo | Java | 实现 Doris RPC UDF 函数服务 | remote-udf-java-demo/README.md |
| remote-udf-cpp-demo | C++ | 实现 Doris RPC UDF 函数服务 | remote-udf-cpp-demo/README.md |
| remote-udf-python-demo | Python | 实现 Doris RPC UDF 函数服务 | remote-udf-python-demo/README.md |
| docs | – | 上述各示例的详细使用说明 | 各子项目 docs/instructions.md |
此外，目录中还包含了 README 未逐一列出的补充示例：java-udf-demo（本地 Java UDF）、java-udf-aes-demo（AES 加解密 UDF）、remote-udaf-java/cpp/python-demo（远程聚合 UDF 服务）、doris-source-demo（Doris SDK 读取示例）、flink-demo-v1.1（面向较早版本 Connector 的示例归档），以及用于统一构建的 pom.xml。
二、flink-demo：基于 Flink Doris Connector 的实时入仓
flink-demo 系列示例从 Flink 框架与 Flink Doris Connector 的视角，讲解如何向 Doris 读写数据，并给出贴近真实业务场景的代码示例。核心类为 org.apache.doris.demo.flink.FlinkConnectorMysqlCDCDemo，完整演示了“MySQL binlog → Flink CDC → Doris 表”的实时同步链路。
2.1 前置准备
2.2 建表：MySQL 源表与 Doris 目标表
先在 MySQL 中创建源表：
CREATE TABLE `test` (
`id` int NOT NULL AUTO_INCREMENT,
`name` varchar(255) DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=19 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci
再在 Doris 中创建目标表（使用 DUPLICATE KEY 明细模型）：
CREATE TABLE `doris_test` (
`id` int NULL COMMENT "",
`name` varchar(100) NULL COMMENT ""
) ENGINE=OLAP
DUPLICATE KEY(`id`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
"replication_num" = "3",
"in_memory" = "false",
"storage_format" = "V2"
);
2.3 注册 Flink 源表与 Doris 结果表
通过 Flink Table API 注册 MySQL CDC 源表：
tEnv.executeSql(
"CREATE TABLE orders (\\n" +
" id INT,\\n" +
" name STRING\\n" +
") WITH (\\n" +
" 'connector' = 'mysql-cdc',\\n" +
" 'hostname' = 'localhost',\\n" +
" 'port' = '3306',\\n" +
" 'username' = 'root',\\n" +
" 'password' = 'zhangfeng',\\n" +
" 'database-name' = 'demo',\\n" +
" 'table-name' = 'test'\\n" +
")");
注册指向 Doris 的结果表（Sink）：
tEnv.executeSql(
"CREATE TABLE doris_test_sink (" +
"id INT," +
"name STRING" +
") " +
"WITH (\\n" +
" 'connector' = 'doris',\\n" +
" 'fenodes' = '10.220.146.10:8030',\\n" +
" 'table.identifier' = 'test_2.doris_test',\\n" +
" 'sink.batch.size' = '2',\\n" +
" 'username' = 'root',\\n" +
" 'password' = ''\\n" +
")");
其中 fenodes 配置为 FE 的 HTTP 端口（示例中为 10.220.146.10:8030），table.identifier 为 库名.表名，sink.batch.size 控制每批写入的行数，username / password 为 Doris 账号。
2.4 执行实时同步
最后执行一条 INSERT INTO 即可将 CDC 流持续写入 Doris：
tEnv.executeSql("INSERT INTO doris_test_sink select id,name from orders");
2.5 源码补充：flink-demo 中的更多场景与底层 Stream Load 封装
从 flink-demo 源码目录 的结构可以看到，该示例还覆盖了更多接入场景：
- kafka/FlinkKafka2Doris.java：Kafka 消息实时写入 Doris；
- hdfs/：自定义 HDFS Source 读取文本后写入 Doris（FlinkReadTextHdfs2Doris、FlinkCustomHdfsSource2Doris）；
- elasticsearch/FlinkReadEs2Doris.java：读取 Elasticsearch 数据同步到 Doris；
- cdc/FlinkOracleCdcDemo.java、JsonDebeziumDeserializationSchema.java：Oracle CDC 场景及 Debezium JSON 反序列化；
- analyze/FlinkDorisConnectorAnalyze.java、FlinkJdbcConnectorAnalyze.java：Connector 与 JDBC 方式对比分析。
其中 DorisStreamLoad.java 是一个可复用的纯 JDK 实现（HttpURLConnection）的 Stream Load 客户端，其核心流程与 Doris 的 Stream Load 协议完全对应：
这套“先打 FE、再跟随 307 重定向到 BE”的交互流程，是所有 Stream Load 客户端（含下文 stream-load-demo）必须处理的核心机制，示例代码中用 setInstanceFollowRedirects(false) 关闭自动跳转、手动读取 Location 的方式实现了对重定向的精确控制。
三、spark-demo：用 Spark 读取 Doris 表
spark-demo 系列示例展示如何使用 Spark 与 Spark Doris Connector 读取 Doris 表数据，同时给出 Java 与 Scala 两种语言的实现，核心类为 org.apache.doris.demo.spark.DorisSparkConnectionExampleJava 与 org.apache.doris.demo.spark.DorisSparkConnectionExampleScala。
注意：与 Flink Connector 类似，Spark Doris Connector 的 jar 文件不在 Maven 中央仓库中，需要单独编译并加入项目 classpath，编译与使用方法参考官方 Spark Doris Connector 文档。
3.1 准备 Doris 表与测试数据
使用任意 MySQL 客户端在 Doris 中建表：
CREATE TABLE `example_table` (
`id` bigint(20) NOT NULL COMMENT "ID",
`name` varchar(100) NOT NULL COMMENT "Name",
`age` int(11) NOT NULL COMMENT "Age"
) ENGINE = OLAP
UNIQUE KEY(`id`)
COMMENT "example table"
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
"replication_num" = "1",
"in_memory" = "false",
"storage_format" = "V2"
);
插入测试数据：
insert into example_table values(1,"xx1",21);
insert into example_table values(2,"xx2",21);
3.2 修改连接配置并运行
在示例类中修改 Doris 连接配置：DORIS_DB、DORIS_TABLE、DORIS_FE_IP、DORIS_FE_HTTP_PORT（默认 8030）、DORIS_FE_QUERY_PORT（默认 9030）、DORIS_USER、DORIS_PASSWORD。运行后应看到如下输出：
+—+—-+—+
| id|name|age|
+—+—-+—+
| 1| xx1| 21|
| 2| xx2| 21|
+—+—-+—+
3.3 源码补充：三种读取方式的实现对照
从 DorisSparkConnectionExampleJava.java 可以看到三种读取 Doris 的方式：
- DataFrame API：sc.read().format("doris")，通过 option("doris.table.identifier", "库.表")、option("doris.fenodes", "FE_IP:8030")、option("user"/"password") 配置后 .load()；
- Spark SQL：CREATE TEMPORARY VIEW … USING doris OPTIONS(…)，再执行 select * from spark_doris；
- JDBC：sc.read().format("jdbc")，URL 为 jdbc:mysql://FE_IP:9030/db?useUnicode=true&characterEncoding=utf-8，指定 dbtable、user、password 后 .load()。
其中方式一、二走 Spark Doris Connector，方式三直接走 MySQL 协议（Doris 兼容 MySQL 协议，查询端口 9030）。此外，spark-demo 源码 中还提供了写入侧示例：demo/hdfs/SparkHdfsToDorisDemo（读取 HDFS 后通过自定义 ForeachPartitionFunction 写入）、demo/kafka/SparkKafkaToDorisDemo，以及基于 Stream Load 的 util/DorisStreamLoad 与 sink/DorisSink，配置常量统一收敛在 constant/SparkToDorisConstants。
四、stream-load-demo：用 Java HttpClient 实现 Stream Load 导入
stream-load-demo 提供 Doris Stream Load 的完整用法示例，包括文件（CSV）与内存数据（JSON）两种导入方式，底层使用 Java HttpClient。
4.1 目标表结构
CREATE TABLE `doris_test_sink` (
`id` int NULL COMMENT "",
`number` int NULL COMMENT "",
`price` DECIMAL(12,2) NULL COMMENT "",
`skuname` varchar(40) NULL COMMENT "",
`skudesc` varchar(200) NULL COMMENT ""
) ENGINE=OLAP
DUPLICATE KEY(`id`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
"replication_num" = "3",
"in_memory" = "false",
"storage_format" = "V2"
);
4.2 文本（CSV）样本数据
10001,12,13.3, test1,this is atest
10002,100,15.3,test2,this is atest
10003,102,16.3,test3,this is atest
10004,120,17.3,test4,this is atest
10005,23,10.3, test5,this is atest
10006,24,112.3,test6,this is atest
10007,26,13.3, test7,this is atest
10008,29,145.3,test8,this is atest
10009,30,16.3, test9,this is atest
100010,32,18.3,test10,this is atest
100011,52,18.3,test11,this is atest
100012,62,10.3,test12,this is atest
4.3 JSON 样本数据
{
"id":556393582,
"number":"123344",
"price":"23.5",
"skuname":"test",
"skudesc":"zhangfeng_test,test"
}
4.4 源码解析：DorisStreamLoader 的导入协议
核心实现为 DorisStreamLoader.java，其关键点如下：
- 导入地址：http://FE_IP:8030/api/{数据库}/{表}/_stream_load（示例中 FE 为 10.220.146.10:8030）；
- 请求方式与认证：HTTP PUT 请求，Authorization 头携带 Basic 认证（用户名:密码 按 UTF-8 Base64 编码）；
- 关键头参数：Expect: 100-continue 提升大包体交互效率；label 使用 UUID.randomUUID() 保证全局唯一（用于导入任务的幂等与查询）；column_separator: "," 指定列分隔符；JSON 导入时追加 format: json；
- 307 重定向处理：通过自定义 DefaultRedirectStrategy 将 isRedirectable 对所有方法返回 true，以支持 FE 返回的 307 跳转（重定向到实际执行导入的 BE），这是 HttpClient 默认策略不处理 PUT 方法重定向时最容易踩的坑；
- 两种实体：文件导入使用 FileEntity（配合 LOAD_FILE_NAME 指定的本地文件路径），JSON 内存导入使用 StringEntity，示例 main 方法默认演示 JSON 导入；
- 结果校验：以 HTTP 状态码是否为 200 判断成功，并打印 Doris 返回的导入结果 JSON（如 Status、NumberTotalRows、NumberLoadedRows 等字段）。
五、spring-jdbc-demo：Spring + MyBatis 提供 Doris 数据 API
spring-jdbc-demo 基于 SpringBoot 与 MyBatis 集成，通过 JDBC 访问 Doris 数据，并通过 REST 接口对外提供数据访问能力（原文：“access doris data through JDBC, and then provide data API access services through rest”）。其表结构与样本数据可直接复用 stream-load-demo 中的定义。
5.1 目录结构与职责
包结构与职责对照如下：
| config | 数据源配置类与应用配置 |
| controller | REST 接口类 |
| datasource | 数据源配置，主要实现多数据源动态切换 |
| domain | 实体 Java Bean |
| mapper | MyBatis 数据访问接口定义 |
| service | 业务服务接口 |
| util | 工具类 |
| DorisApplication | SpringBoot 启动类 |
对应配置包括：application.yml（SpringBoot 主配置）、application-druid.yml（数据库连接池连接参数）、MyBatis 相关参数及 mapper.xml 数据访问映射文件。
从源码看，datasource 下的 DynamicDataSource、DynamicDataSourceContextHolder 与 DataSourceType 实现了基于 ThreadLocal 的多数据源动态路由，可在运行期按需切换到不同 Doris 库或集群；controller/DemoController.java 对外暴露 REST 端点，经 service → mapper（MyBatis）最终以 JDBC 方式查询 Doris。
5.2 接口调用
启动 SpringBoot 后，在浏览器地址栏访问 RESTful URL：
- 查询列表：访问 localhost:8080/rest/demo/skulist，返回 SKU 列表数据，效果如下图所示：
- 按 ID 查询：按主键查询单条记录，效果如下图所示：
该示例说明了一个通用接入模式：Doris 兼容 MySQL 协议，任何 MySQL 生态的中间件（Spring Data、MyBatis、MyBatis-Plus、JDBC 连接池等）都可以直接对接 Doris，将 Doris 作为分析型数据源快速嵌入既有 Java 服务体系。
六、RPC UDF 函数服务：Java / C++ / Python 三种实现
RPC UDF（远程 UDF）是 Doris 的扩展机制：用户将 UDF 实现为独立的 RPC 函数服务进程，Doris BE 通过 gRPC 调用该服务完成计算。三个示例（remote-udf-java-demo、remote-udf-cpp-demo、remote-udf-python-demo）演示了同一套协议在三种语言下的实现，proto 定义统一存放在各目录的 proto/function_service.proto 与 proto/types.proto 中（与 gensrc/proto/function_service.proto 对应），核心 RPC 接口包括 fnCall（执行函数）、checkFn（校验函数签名）、handShake（握手）。
6.1 Java 实现：编译与运行
mvn package
java -jar target/remote-udf-java-demo-jar-with-dependencies.jar 9000
其中 9000 是服务监听的端口。
从 FunctionServiceImpl.java 可以看到 Java 侧的实现模式：
- fnCall 分发：根据 request.getFunctionName() 分发到具体函数——add_int_two（两列 INT 逐行相加）、add_int_one（INT 每行加一）、add_string（字符串追加 _rpc_test），结果通过 PFunctionCallResponse 返回，其中 PValues 携带类型元信息（PGenericType.TypeId.INT32 / STRING）与批量列式数据；
- checkFn 校验：校验函数输入参数个数，例如 add_int_two 必须恰好 2 个输入，否则返回非 0 状态码；
- handShake 握手：回显 hello 字段完成连接握手。
这种“函数名分发 + 列式批量入参/出参 + 显式类型声明”的设计与 Doris 向量化执行引擎天然契合——PValues 一次性携带一整列数据，避免逐行 gRPC 调用开销。
6.2 C++ 实现：编译与运行
make clean && make all
./function_server 9000
其中 9000 是服务监听的端口。C++ 版本同样基于 proto/function_service.proto 生成 gRPC 桩代码，主程序为 cpp_function_service_demo.cpp，用 Makefile 完成 protoc 代码生成与链接构建。
6.3 Python 实现：编译与运行
Python 版本需要先用 grpcio-tools 从 proto 生成代码，再启动服务：
pip install grpcio-tools
python -m grpc_tools.protoc -Iproto –python_out=. –grpc_python_out=. proto/function_service.proto && python -m grpc_tools.protoc -Iproto –python_out=. proto/types.proto
python function_server_demo.py 9000
其中 9000 是服务监听的端口，服务端主逻辑位于 function_server_demo.py。
6.4 同类扩展：远程聚合 UDF（UDAF）
目录中还提供了 remote-udaf-java-demo、remote-udaf-cpp-demo、remote-udaf-python-demo 三个聚合函数版本，复用同一套 function_service.proto / types.proto，从目录结构与示例文件命名可以推断其用于演示如何实现 Doris 的远程 UDAF（聚合函数）服务，作为 UDF 示例在聚合计算场景的延伸。
七、更多示例与构建方式
7.1 本地 Java UDF
java-udf-demo/README.md 提供本地 Java UDF 的示例，编译命令为 mvn clean package。源码中 AddOne.java 演示“入参加一”的标量函数，DateCaseTest.java 演示日期类型参数的处理。与之配套的 java-udf-aes-demo 从类名（AESEncrypt、AESDecrypt、AESUtil）可以推断其演示基于 AES 算法的加解密 UDF 实现。
7.2 Doris SDK 读取示例
doris-source-demo 中提供了 org.apache.doris.sdk.DorisReaderExample，从包名与类名可以推断该示例演示通过 Doris 提供的 SDK 方式读取 Doris 数据，可配合其 README.md 查看说明。
7.3 早期版本示例归档
flink-demo-v1.1 面向较早版本的 Flink Doris Connector，从示例类名（DorisFlinkConnectorDemoV1、Cdc2DorisSQLDemo、Kafka2DorisDataStreamDemo、DatabaseFullSync、DatabaseFullDelSync 等）可以推断其覆盖了 CDC 转 SQL 接入、Kafka DataStream 写入以及整库全量/增量同步等典型场景，适合作为版本演进时的参照。
7.4 构建方式
整个 doris-demo 由根 pom.xml 聚合管理，各子示例拥有独立的 pom.xml，可按需单独进入对应子目录执行 Maven 构建（如 mvn clean package）。各示例的详细说明统一汇总在各子项目的 docs/instructions.md 中。
八、小结
doris-demo 用最小化的示例代码覆盖了 Doris 生态接入的四大关键路径：连接器路径（Flink / Spark Doris Connector 的读写与 CDC）、导入路径（Stream Load 的文件与 JSON 导入，含 307 重定向与 Basic Auth 协议细节）、服务化路径（Spring + MyBatis + JDBC 快速对外提供查询 API）、扩展路径（Java / C++ / Python 三种语言的 RPC UDF 服务，以及本地 Java UDF 与 UDAF）。开发者可以按图索骥：先对照各 docs/instructions.md 完成建表与环境准备，再参考对应源码类调整连接参数即可落地运行，进而结合本文的源码剖析理解 Stream Load 重定向、列式 RPC 传参等底层协议设计。
赞
【免费下载链接】doris
Apache Doris is an easy-to-use, high performance and unified analytics database.
项目地址：
https://gitcode.com/gh_mirrors/dori/doris
点击查看 免费下载
相关推荐
轻松掌握macOS农历工具：菜单栏应用终极指南
如何快速配置NeverSink过滤器：流放之路2玩家的终极效率指南
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
