首页 > 软件编程 > java > SpringBoot3 Calcite多数据源查询

SpringBoot3集成Calcite多数据源查询的实战示例小结

2026-02-09 10:22:02 作者：暴躁代码

本文介绍了Spring Boot 3集成Apache Calcite实现多数据源查询的实战指南,本文通过实例代码给大家讲解的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧

前言：跨库查询的痛，谁懂？

凌晨两点被告警电话叫醒，订单查询接口超时，客服群炸开了锅。排查了一圈才发现问题的根源：订单数据在 MySQL，用户信息在 PostgreSQL，行为画像在 MongoDB，访问日志存在 Hive。四套 DAO 层相互牵制，改动任何一个字段就像推倒多米诺骨牌一样引发连锁反应。

企业项目一旦数据源增多，"多数据源管理"很快就演变成维护黑洞。每新增一个数据源，就要新建一套 DAO 配置、数据源连接池、事务管理。复用率低不说，还把业务逻辑和数据访问层死死绑在一起。最要命的是，不同数据源的字段命名规范不统一，user_level、customer_level、user_grade 混着用，改一个字段名，订单系统、用户系统、风控系统、报表系统全线报错，谁改谁背锅。

传统方案要么做全量数据同步到数仓，要么写一堆分布式事务代码，要么在内存里手动 JOIN。结果就是：延迟不可避免，查询结果拿到的往往是旧数据快照；性能是硬伤，MySQL 擅长点查和小范围索引扫描，MongoDB 天生文档检索，Hive 批量扫描效率高，Kafka 流式处理，一刀切的内存聚合方式很快就把 JVM 打爆了，查询延迟像筛子一样抖个不停。

一、重新认识 Apache Calcite：不只是数据库，更是查询大脑

很多人第一次听到 “Apache Calcite”，直觉反应是"又一个数据库"。这个理解完全错了。

Calcite 本质上是一个动态数据管理框架，专注于提供 SQL 解析、查询优化和跨数据源连接的基础能力，但不涉及数据的存储和处理。这种设计哲学让它具备了极其灵活的适配能力。

它主要做四件事：

1. SQL 解析与验证

将用户提交的 SQL 语句解析为抽象语法树（AST），并进行语义分析，验证表名、列名是否存在，数据类型是否匹配等。就像把一句自然语言翻译成结构化的指令树。

2. 查询优化

这是 Calcite 的核心杀手锏。它提供基于规则优化（RBO）和基于代价优化（CBO）两种策略：

规则优化：通过预设规则重写查询，比如谓词下推、投影裁剪、JOIN 重排
代价优化：根据统计信息估算不同执行计划的成本，选择最优方案

3. 数据源适配

Calcite 通过适配器（Adapter）机制连接各种数据源，包括：

关系型数据库：MySQL、PostgreSQL、Oracle、SQL Server
NoSQL 数据库：MongoDB、Cassandra、Redis
文件系统：CSV、JSON、Parquet、Excel
大数据引擎：Hive、Spark、Kafka

甚至还可以自定义适配器，让任何数据源都能接入。

4. 跨数据源查询

能够连接不同类型的数据源，通过适配器统一抽象不同数据源的操作，将查询分解为各数据源可处理的子查询，然后合并结果。

一句话总结：Calcite 把"数据在哪"和"怎么查"彻底拆开了。你写一句标准 SQL，它负责解析、优化、拆分，最终把查询路由到各个数据源执行。

很多熟悉的系统都在用 Calcite：Flink SQL 用它做解析与优化，Hive 的 CBO 用它打底，Drill、Kylin、Druid 都接入了它的能力。

二、Spring Boot 3 集成 Calcite 实战指南

2.1 核心依赖引入

第一步是添加 Maven 依赖。在 pom.xml 中引入 Calcite 核心包、对应数据源的适配器，以及 MyBatis Plus 的核心依赖。

<!-- Calcite 核心依赖 -->
<dependency>
    <groupId>org.apache.calcite</groupId>
    <artifactId>calcite-core</artifactId>
    <version>1.36.0</version>
</dependency>
<!-- MySQL 适配器 -->
<dependency>
    <groupId>org.apache.calcite</groupId>
    <artifactId>calcite-mysql</artifactId>
    <version>1.36.0</version>
</dependency>
<!-- MongoDB 适配器 -->
<dependency>
    <groupId>org.apache.calcite</groupId>
    <artifactId>calcite-mongodb</artifactId>
    <version>1.36.0</version>
</dependency>
<!-- MyBatis Plus 核心依赖（需要适配 Spring Boot 3） -->
<dependency>
    <groupId>com.baomidou</groupId>
    <artifactId>mybatis-plus-boot-starter</artifactId>
    <version>3.5.5</version>
</dependency>
<!-- 数据源连接池 -->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>druid-spring-boot-starter</artifactId>
    <version>1.2.20</version>
</dependency>

三个避坑点必须强调：

Calcite 所有组件版本必须统一。核心包和适配器版本要一致，否则容易出现类加载异常。
MyBatis Plus 要选适配 Spring Boot 3 的版本。必须是 3.5.3 及以上版本，旧版本不支持。
一定要加连接池依赖。MyBatis Plus 需要连接池支持才能正常管理 Calcite 数据源。

2.2 编写 Calcite 模型文件

模型文件是 Calcite 识别数据源的关键，通常使用 JSON 格式，放在 resources 目录下，命名为 calcite-model.json。

下面是一个适配 MySQL 和 MongoDB 双数据源的示例：

{
  "version": "1.0",
  "defaultSchema": "ecommerce",
  "schemas": [
    {
      "name": "ecommerce",
      "type": "custom",
      "factory": "org.apache.calcite.adapter.jdbc.JdbcSchema$Factory",
      "operand": {
        "jdbcUrl": "jdbc:mysql://localhost:3306/ecommerce_order?useSSL=false&serverTimezone=UTC",
        "username": "root",
        "password": "123456",
        "driver": "com.mysql.cj.jdbc.Driver"
      }
    },
    {
      "name": "user_mongo",
      "type": "custom",
      "factory": "org.apache.calcite.adapter.mongodb.MongoSchema$Factory",
      "operand": {
        "host": "localhost",
        "port": 27017,
        "database": "user_db",
        "collection": "user_info"
      }
    }
  ]
}

关键配置说明：

defaultSchema：默认查询的 schema，可省略。如果省略，查询时需要指定 schema 名称（如 ecommerce.order、user_mongo.user_info）。
factory：对应数据源的适配器工厂类。Calcite 已为主流数据源提供现成工厂，自定义数据源需要实现自己的工厂类。
operand：数据源连接参数，根据数据源类型不同配置不同参数（如 MySQL 的 jdbcUrl、MongoDB 的 host/port）。

2.3 Spring Boot 集成 Calcite + MyBatis Plus 核心配置

这一步是集成的核心，主要分两步走：

配置好 Calcite 数据源
让 MyBatis Plus 使用这个数据源，并配置 Mapper 扫描、分页插件等基础参数

import com.baomidou.mybatisplus.annotation.DbType;
import com.baomidou.mybatisplus.extension.plugins.MybatisPlusInterceptor;
import com.baomidou.mybatisplus.extension.plugins.inner.PaginationInnerInterceptor;
import org.apache.calcite.jdbc.CalciteConnection;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.util.Properties;
@Configuration
@MapperScan(basePackages = "com.example.calcite.mapper")
public class CalciteMybatisPlusConfig {
    // 1. 配置 Calcite 数据源
    @Bean
    public DataSource calciteDataSource() throws Exception {
        Properties props = new Properties();
        props.setProperty("model", "classpath:calcite-model.json");
        Connection connection = DriverManager.getConnection("jdbc:calcite:", props);
        CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class);
        return calciteConnection.getDataSource();
    }
    // 2. 配置 MyBatis Plus 的 SqlSessionFactory，指定使用 Calcite 数据源
    @Bean
    public SqlSessionFactory sqlSessionFactory(DataSource calciteDataSource) throws Exception {
        MybatisSqlSessionFactoryBean sessionFactory = new MybatisSqlSessionFactoryBean();
        // 注入 Calcite 数据源
        sessionFactory.setDataSource(calciteDataSource);
        // 配置 Mapper.xml 文件路径
        sessionFactory.setMapperLocations(
            new PathMatchingResourcePatternResolver().getResources("classpath:mapper/*.xml")
        );
        // 配置 MyBatis Plus 全局参数
        org.apache.ibatis.session.Configuration configuration = 
            new org.apache.ibatis.session.Configuration();
        configuration.setMapUnderscoreToCamelCase(true); // 下划线转驼峰
        sessionFactory.setConfiguration(configuration);
        // 注入 MyBatis Plus 插件
        sessionFactory.setPlugins(mybatisPlusInterceptor());
        return sessionFactory.getObject();
    }
    // 3. MyBatis Plus 分页插件
    @Bean
    public MybatisPlusInterceptor mybatisPlusInterceptor() {
        MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
        interceptor.addInnerInterceptor(
            new PaginationInnerInterceptor(DbType.MYSQL) // 适配 Calcite 兼容的 MySQL 语法
        );
        return interceptor;
    }
    // 4. 配置事务管理器
    @Bean
    public PlatformTransactionManager transactionManager(DataSource calciteDataSource) {
        return new DataSourceTransactionManager(calciteDataSource);
    }
}

核心逻辑梳理：

先通过 Calcite 创建统一的数据源，再把它注入到 MyBatis Plus 的 SqlSessionFactory 里。这样一来，后续写代码完全是 MyBatis Plus 的熟悉风格，无论是 Mapper 接口还是 XML 映射文件，都能直接用，跨数据源查询的复杂逻辑全交给 Calcite 处理。

2.4 核心查询实现

定义实体类

使用 Lombok 注解定义 VO 类，包含订单和用户信息：

import lombok.Data;
@Data
public class UserOrderVO {
    // 订单信息
    private String orderId;
    private String orderTime;
    private Double amount;
    // 用户信息
    private String userName;
    private String phone;
    private String userId;
}

定义 Mapper 接口

继承 BaseMapper 获得 MyBatis Plus 基础 CRUD 能力，使用 @Select 注解编写跨数据源关联 SQL：

import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Select;
import java.util.List;
@Mapper
public interface UserOrderMapper extends BaseMapper<UserOrderVO> {
    // 注解方式编写跨数据源关联 SQL
    @Select("SELECT " +
            "  o.order_id AS orderId, " +
            "  o.order_time AS orderTime, " +
            "  o.amount, " +
            "  u.user_name AS userName, " +
            "  u.phone " +
            "FROM ecommerce.orders o " +
            "JOIN user_mongo.user_info u ON o.user_id = u.user_id " +
            "WHERE o.user_id = #{userId}")
    List<UserOrderVO> queryUserOrderByUserId(String userId);
}

如果使用 XML 方式：

在 resources/mapper/UserOrderMapper.xml 中编写：

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" 
    "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.example.calcite.mapper.UserOrderMapper">
    <select id="queryUserOrderByUserIdWithXml" resultType="com.example.calcite.entity.UserOrderVO">
        SELECT
            o.order_id AS orderId,
            o.order_time AS orderTime,
            o.amount,
            u.user_name AS userName,
            u.phone
        FROM ecommerce.orders o
        JOIN user_mongo.user_info u ON o.user_id = u.user_id
        WHERE o.user_id = #{userId}
    </select>
</mapper>

编写 Service 层

继承 ServiceImpl 实现业务逻辑：

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.stereotype.Service;
import java.util.List;
@Service
public class UserOrderService extends ServiceImpl<UserOrderMapper, UserOrderVO> {
    public List<UserOrderVO> getUserOrderByUserId(String userId) {
        // 直接调用 Mapper 接口方法
        return this.baseMapper.queryUserOrderByUserId(userId);
    }
    // 如果使用 XML 方式：
    public List<UserOrderVO> getUserOrderByUserIdWithXml(String userId) {
        return this.baseMapper.queryUserOrderByUserIdWithXml(userId);
    }
}

编写 Controller 层

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
@RestController
public class CrossDataSourceQueryController {
    @Autowired
    private UserOrderService userOrderService;
    @GetMapping("/user/order/{userId}")
    public List<UserOrderVO> queryUserOrder(@PathVariable String userId) {
        // 调用 Service 方法，返回跨数据源查询结果
        return userOrderService.getUserOrderByUserId(userId);
    }
}

三个关键点：

实体类字段要和查询结果列名对应。使用别名适配下划线转驼峰更省心（如 o.order_id AS orderId）。
Mapper 接口继承 BaseMapper 后，MyBatis Plus 的分页、条件构造器这些功能都能正常使用。
SQL 查询要带 schema 前缀。比如 ecommerce.orders、user_mongo.user_info，明确指定数据源。

三、经典使用场景深度解析

3.1 多系统数据融合查询

场景痛点：

大型企业里数据分散在不同系统，订单系统用 MySQL，用户系统用 MongoDB 存行为数据，库存系统用 PostgreSQL。传统做法需要分别调用三个系统接口，在内存中手动整合数据，效率低且容易出错。一旦某个系统字段变更，所有调用方都要改。

Calcite 解决方案：

用 Calcite 分别适配三个数据源后，只要写一套标准 SQL 就能实现跨数据源关联查询。业务层完全不用管数据存在哪，专注核心业务逻辑。

价值收益：

开发效率提升 50% 以上。不用写重复的接口调用和数据整合代码。
Calcite 的查询优化器会自动优化关联逻辑。先把时间过滤推给 MySQL 做索引扫描，再把用户 ID 集合下推给 MongoDB 做 IN 查询，最后在本地完成 JOIN 和聚合。
查询效率也能跟上。谓词下推、投影裁剪、JOIN 重排这些优化都是自动的，不用手动抠。

3.2 实时数据与离线数据联动查询

场景需求：

运营需要实时查看今日订单加近 30 天历史订单的汇总数据。实时订单数据存在 Kafka 里，历史订单数据存在 Hive 里。

传统方案：

需要开发两套查询逻辑，分别从 Kafka 和 Hive 拉数据，再在内存中合并。开发成本高，而且数据同步有延迟，合并后的结果未必是实时的。

Calcite 解决方案：

用 Calcite 的 Kafka 适配器和 Hive 适配器，把实时流数据和离线数据放到同一个查询体系里。写一条 SQL 就能实现实时加离线数据的联合查询：

SELECT 
  product_id,
  SUM(amount) AS total_amount
FROM (
  -- 实时数据（Kafka）
  SELECT product_id, amount FROM realtime.orders WHERE dt = CURRENT_DATE
  UNION ALL
  -- 离线数据（Hive）
  SELECT product_id, amount FROM hive.orders WHERE dt >= CURRENT_DATE - INTERVAL '30' DAY
) combined
GROUP BY product_id

价值收益：

省去了数据同步成本。不用把 Kafka 数据实时同步到 Hive，也不用把 Hive 数据预聚合出来。
兼顾实时性和准确性。Kafka 的实时数据 + Hive 的历史数据，联合查询结果既有实时性又有完整性。

3.3 自定义数据源适配

场景需求：

企业里有很多 CSV、Excel、Parquet 格式的文件数据，需要集成到业务系统中查询。

传统方案：

先把这些文件导入数据库（如 MySQL），然后再提供查询接口。数据迁移成本高，而且数据更新后需要重新导入。

Calcite 解决方案：

Calcite 内置了文件适配器，支持直接查询这些文件数据，根本不用导入数据库。结合 Spring Boot 3 的文件上传功能，还能实现文件上传后直接用 SQL 查询的需求。

模型配置示例：

{
  "name": "files",
  "type": "custom",
  "factory": "org.apache.calcite.adapter.csv.CsvSchemaFactory",
  "operand": {
    "directory": "/data/files",
    "flavor": "mysql"
  }
}

查询示例：

-- 直接查询 CSV 文件
SELECT * FROM files.sales_data WHERE region = 'East'
-- 与其他数据源关联
SELECT 
  f.product_id,
  p.product_name,
  f.sales_amount
FROM files.sales_data f
JOIN ecommerce.products p ON f.product_id = p.id

四、避坑指南：集成注意事项

4.1 版本一致性

问题：

Calcite 核心依赖和各数据源适配器的版本必须一致，不然很容易出现类加载异常。

避坑技巧：

推荐使用 Calcite 1.36.0 版本。这个版本稳定性好，适配器齐全。
不要混用不同版本。比如 calcite-core 1.36.0 + calcite-mysql 1.35.0，这种组合极易出问题。
用 mvn dependency:tree 检查依赖冲突。

4.2 模型文件配置规范

问题：

Schema 名称、表名要清晰，别重复。数据源的地址、端口、账号密码这些连接参数一定要准确。

避坑技巧：

按业务域划分 Schema。比如 retail.orders、crm.users，一眼能看懂归属。
表名使用有意义的名称。避免用 table1、table2 这种模糊命名。
连接参数做好环境隔离。开发、测试、生产环境用不同的配置文件。

4.3 数据源性能考虑

问题：

跨数据源查询的性能取决于最慢的那个数据源。如果 MySQL 查询很快，但 MongoDB 慢，整体查询还是会被拖慢。

避坑技巧：

确保每个数据源自身性能没问题。定期检查慢查询，优化索引。
合理使用谓词下推。Calcite 会自动把过滤条件推到数据源执行，但要检查执行计划，确保推下去了。
对小表考虑广播到内存，避免大表大表硬碰硬。

4.4 SQL 方言差异

问题：

不同数据源支持的 SQL 语法有差异。比如 grouping sets、窗口函数、子查询某些源不完全支持。

避坑技巧：

查询前先测试。开发阶段用 Calcite 的 EXPLAIN 功能看执行计划。
不支持的语法要降级实现。比如某个源不支持窗口函数，就改用子查询。
定期查看 Calcite 和各数据源的官方文档，了解语法支持情况。

五、优化小技巧：让查询更快更稳

5.1 启用 Calcite 缓存

元数据缓存：

避免每次查询都重新加载 Schema 结构，减少重复解析和元数据查询的时间。

props.setProperty("calcite.metadataCacheSize", "1000");

查询计划缓存：

对相同 SQL 查询，缓存执行计划，提升重复查询效率。

props.setProperty("calcite.parser.factory", 
    "org.apache.calcite.sql.parser.impl.SqlParserImpl#FACTORY");

5.2 优化 SQL 写法

能下推就下推：

尽量避免复杂的多表关联在 Calcite 层执行，把过滤条件下推到数据源。

示例：

-- 不好的写法：先全量扫描再过滤
SELECT * FROM ecommerce.orders o JOIN user_mongo.users u 
  ON o.user_id = u.user_id 
  WHERE o.create_time >= '2025-01-01'
-- 好的写法：时间过滤下推到 MySQL
SELECT * FROM (
  SELECT * FROM ecommerce.orders WHERE create_time >= '2025-01-01'
) o 
JOIN user_mongo.users u ON o.user_id = u.user_id

选择必要字段：

避免 SELECT *，只查询需要的字段，减少数据传输量。

5.3 自定义优化规则

如果是特别复杂的业务场景，可以自己实现 RelOptRule 接口，写自定义的查询优化规则。

示例场景：

强制某些过滤条件先执行，或者把某个维表标成可广播。

代码示例：

import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.rules.RelOptRule;
public class CustomRule extends RelOptRule {
    public CustomRule() {
        super(operand(FilterRel.class, any()));
    }
    @Override
    public boolean matches(RelOptRuleCall call) {
        // 自定义匹配逻辑
        return true;
    }
    @Override
    public void onMatch(RelOptRuleCall call) {
        // 自定义优化逻辑
        FilterRel filter = call.rel(0);
        // ... 重写查询计划
    }
}

5.4 开启 EXPLAIN 分析

定期分析慢查询的执行计划，找出性能瓶颈：

EXPLAIN PLAN FOR 
SELECT * FROM ecommerce.orders o 
JOIN user_mongo.user_info u ON o.user_id = u.user_id 
WHERE o.user_id = '123456'

重点关注：

拉回来的行数
下推的过滤条件
JOIN 顺序
是否使用了索引

六、边界与选型：什么时候不该用 Calcite

6.1 不适合的场景

强事务 OLTP 场景：

Calcite 不适合替代 OLTP 数据库的核心业务场景。对于强事务需求，如转账扣减、库存冻结，仍需使用传统数据库，跨库分布式事务别硬凑。

大规模分布式计算：

如果要的是超大规模分布式计算，直接开 Trino、Presto 集群。Calcite 更像"查询大脑"，把解析和优化做好，执行靠接出来的源，或者你自己写执行器。

高频点查场景：

对于高频单表点查，直接用数据源自身的驱动可能更高效，没必要经过 Calcite 这一层。

6.2 适合的场景

数据中台：作为统一查询层
多系统整合：跨部门、跨系统的数据融合
实时分析：流批一体化分析
数据虚拟化：免 ETL，直接查询原始数据源

6.3 与其他方案的对比

方案	优点	缺点	适用场景
Calcite	嵌入式、自定义规则强、轻量级	分布式能力弱	Spring Boot 应用内集成、规则定制化需求强
Trino/Presto	分布式、大规模计算	学习曲线陡、集群运维成本高	数据仓库、大规模分析
ETL	稳定、数据一致性好	实时性差、开发量大	离线数仓、数据同步为主
BI 平台	易用、可视化强	依赖工具、定制化受限	业务分析、报表展示

七、团队协作与治理

7.1 统一 SQL 编码规范

强制使用 Schema.table 格式访问表，避免歧义。
禁用 SELECT *，只查询必要字段。
统一字段命名规范，下划线和驼峰明确约定。

7.2 建立查询评审机制

复杂查询需经过架构评审。
评审重点关注：查询逻辑、执行计划、数据量预估。
评审通过才能上线。

7.3 监控与告警

查询性能监控：对慢查询（超过 1 秒）进行告警。
跨源查询数据量监控：记录每次查询涉及的行数，及时发现异常。
连接池监控：监控各数据源连接池使用情况，避免连接耗尽。

7.4 定期清理与维护

跨域配置按季度清理：没用的数据源配置就摘掉，避免"历史债务"不断叠加。
元数据定期更新：数据源表结构变更后，及时更新模型文件。
规则库维护：自定义优化规则要定期review，避免规则过时。

八、总结

Spring Boot 3 集成 Apache Calcite，最大的变化是：

业务写 SQL，不用管数据在哪
适配器管"去哪儿查"，自动路由到对应数据源
优化器管"怎么查得更省"，自动下推、自动优化
工程师把精力放回业务规则，而不是堆 DAO、抠连接

它不是万能钥匙，但在多源并存的公司里，它确实把"跨库地狱"挪走了一大半。

用对地方，别把它硬拽去当分布式 OLTP，剩下的，交给监控、规范、演练和一点点耐心。

技术选型的本质，不是找最先进的工具，而是找最合适的工具。

参考资料：

Apache Calcite 官方文档：https://calcite.apache.org/
Spring Boot 3 官方文档
MyBatis Plus 官方文档
Apache Flink、Hive 等大数据系统的 Calcite 集成案例

到此这篇关于SpringBoot3集成Calcite多数据源查询实战笔记-7407466045的文章就介绍到这了,更多相关SpringBoot3 Calcite多数据源查询内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！