SpringBoot與Calcite整合，實現多數據源統一查詢系統
作者：Java知識日歷
最近，接到一個電商系統的兼職小單，其中訂單信息存儲在MySQL數據庫，而用戶信息存儲在PostgreSQL數據庫。客戶那邊想有一個統一查詢接口，可以通過SQL查詢同時獲取這兩個數據源的信息。
最近，接到一個電商系統的兼職小單，其中訂單信息存儲在MySQL數據庫，而用戶信息存儲在PostgreSQL數據庫?？蛻裟沁呄胗幸粋€統一查詢接口，可以通過SQL查詢同時獲取這兩個數據源的信息。
為什么選擇Apache Calcite？
簡化開發流程
- 抽象層次高: Apache Calcite 提供了高層次的抽象，使得開發者可以專注于業務邏輯，而不必處理底層的數據庫連接和查詢執行細節。
- 減少重復工作: 通過使用Calcite，可以避免重復造輪子，節省開發時間和成本。
強大的SQL解析和優化能力
- SQL標準支持: Apache Calcite 支持多種SQL方言（如MySQL、PostgreSQL等），可以無縫地處理不同數據庫的SQL語句。
- 查詢優化: 內置的查詢優化器可以根據不同的數據源特性進行智能優化，提高查詢性能。
靈活性和可擴展性
- 自定義模式和表: 可以通過編程方式動態地添加和管理多個數據源，每個數據源可以有不同的模式和表結構。
- 插件機制: 支持各種插件，可以根據需求靈活擴展功能，例如自定義函數、聚合操作等。
高性能
- 內存計算: Apache Calcite 支持內存中的數據處理，減少了I/O開銷，提高了查詢速度。
- 分布式計算: 雖然本項目主要關注單機版實現，但Apache Calcite也可以擴展到分布式環境中，支持大規模數據集的處理。
集成性強
- 與其他工具集成: 支持與其他大數據工具和技術棧（如Apache Flink、Presto等）集成，形成完整的數據分析解決方案。
哪些公司使用了Apache Calcite？
- Google 在其內部的一些數據處理系統中使用 Apache Calcite，特別是在需要高性能和靈活性的場景下。
- IBM 在其數據倉庫和分析解決方案中使用 Apache Calcite，以提高查詢性能和靈活性。
- Intel 使用 Apache Calcite 來支持其大數據分析工具和解決方案，特別是在內存計算方面。
- Alibaba Cloud: 阿里巴巴云在其大數據平臺中使用 Apache Calcite 提供強大的查詢優化和執行能力。
- MaxCompute (ODPS): 阿里巴巴的大規模數據計算服務 MaxCompute 使用 Calcite 進行 SQL 查詢處理。
- Elasticsearch 的某些高級功能，如 Kibana 中的復雜查詢，依賴于 Apache Calcite 進行 SQL 解析和優化。
- Netflix 使用 Apache Calcite 來構建其內部的數據虛擬化層，支持復雜的查詢和數據分析需求。
- Microsoft 在其一些大數據產品和服務中使用 Apache Calcite，例如 Azure Synapse Analytics。
- Teradata 使用 Apache Calcite 來增強其數據庫系統的查詢優化和執行性能。
- Uber 使用 Apache Calcite 來處理其龐大的數據集，并支持復雜的查詢和數據分析需求。
應用場景
數據虛擬化
- 虛擬數據層: 創建一個虛擬的數據層，將分散在不同系統中的數據集中起來，提供統一的視圖。
- 動態數據源管理: 動態地添加和管理數據源，支持靈活的數據架構設計。
商業智能 (BI) 工具
- 報表生成: 作為 BI 工具的核心組件，支持復雜的報表生成和數據分析。
- 自助服務分析: 提供自助服務分析功能，允許非技術人員進行數據探索和分析。
機器學習與人工智能
- 特征工程: 在機器學習管道中使用 Calcite 進行特征提取和數據準備。
- 模型訓練: 結合其他 AI 框架，利用 Calcite 進行大規模數據集的查詢和處理。
多數據源查詢
- 統一接口訪問多個數據庫: 允許用戶通過單一接口查詢存儲在不同數據庫（如 MySQL、PostgreSQL、Oracle 等）中的數據。
- 聯合查詢: 支持跨數據源的復雜 SQL 查詢，例如從不同的數據庫中獲取相關聯的數據。
大數據平臺集成
- 與 Hadoop 生態系統集成: 與 Hive、HBase、Druid 等大數據工具結合，提供統一的查詢接口。
- 流處理與批處理: 支持 Apache Flink 和 Apache Beam 等流處理框架，實現實時數據分析。
嵌入式數據庫
- 輕量級數據庫引擎: 提供一個輕量級的 SQL 引擎，適用于嵌入式應用程序和內存數據庫。
- 內存計算: 利用內存計算加速查詢性能，適合需要快速響應的應用場景。
數據湖解決方案
- 統一元數據管理: 提供統一的元數據管理和查詢接口，方便數據湖的建設和維護。
- 多樣化數據格式支持: 支持多種數據格式（如 JSON、Parquet、ORC 等），滿足不同類型的數據存儲需求。
代碼實操
<dependencies>
<!-- Spring Boot Starter Web -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- Apache Calcite Core -->
<dependency>
<groupId>org.apache.calcite</groupId>
<artifactId>calcite-core</artifactId>
<version>1.32.0</version>
</dependency>
<!-- HikariCP Connection Pool -->
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP</artifactId>
</dependency>
<!-- MySQL Connector Java -->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<scope>runtime</scope>
</dependency>
<!-- PostgreSQL JDBC Driver -->
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId>
<scope>runtime</scope>
</dependency>
<!-- Test Dependencies -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>application.yml
spring:
datasource:
order-db:
url:jdbc:mysql://localhost:3306/order_db?useSSL=false&serverTimezone=UTC
username:root
password:root
driver-class-name:com.mysql.cj.jdbc.Driver
user-db:
url:jdbc:postgresql://localhost:5432/user_db
username:postgres
password:postgres
driver-class-name:org.postgresql.Driver
jpa:
show-sql:true
hibernate:
ddl-auto:update
properties:
hibernate:
dialect:org.hibernate.dialect.MySQL8Dialect數據源配置
package com.example.multids.config;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.sql.DataSource;
@Configuration
publicclass DataSourceConfig {
@Bean(name = "mysqlDataSource")
public DataSource mysqlDataSource() {
// 配置MySQL數據源
HikariConfig config = new HikariConfig();
config.setJdbcUrl("jdbc:mysql://localhost:3306/order_db?useSSL=false&serverTimezone=UTC");
config.setUsername("root");
config.setPassword("root");
returnnew HikariDataSource(config);
}
@Bean(name = "postgresDataSource")
public DataSource postgresDataSource() {
// 配置PostgreSQL數據源
HikariConfig config = new HikariConfig();
config.setJdbcUrl("jdbc:postgresql://localhost:5432/user_db");
config.setUsername("postgres");
config.setPassword("postgres");
returnnew HikariDataSource(config);
}
}自定義數據源工廠
package com.example.multids.factory;
import com.example.multids.schema.MySchemas;
import org.apache.calcite.adapter.jdbc.JdbcSchema;
import org.apache.calcite.adapter.java.ReflectiveSchema;
import org.apache.calcite.config.Lex;
import org.apache.calcite.jdbc.CalciteConnection;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
publicclass DataSourceFactory {
public static CalciteConnection createConnection(DataSource mysqlDataSource, DataSource postgresDataSource) throws SQLException {
// 定義Calcite模型JSON字符串
String modelJson = "{\n" +
" \"version\": \"1.0\",\n" +
" \"defaultSchema\": \"my_schemas\",\n" +
" \"schemas\": [\n" +
" {\n" +
" \"name\": \"my_schemas\",\n" +
" \"type\": \"custom\",\n" +
" \"factory\": \"" + ReflectiveSchema.Factory.class.getName() + "\",\n" +
" \"operand\": {\n" +
" \"class\": \"" + MySchemas.class.getName() + "\"\n" +
" }\n" +
" }\n" +
" ]\n" +
"}";
// 創建Calcite連接
Connection connection = DriverManager.getConnection("jdbc:calcite:model=" + modelJson);
CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class);
// 獲取根模式并添加子模式
SchemaPlus schema = calciteConnection.getRootSchema().getSubSchema("my_schemas");
schema.add("orders", JdbcSchema.create(calciteConnection.getRootSchema(), "orders", mysqlDataSource, null, Lex.MYSQL));
schema.add("users", JdbcSchema.create(calciteConnection.getRootSchema(), "users", postgresDataSource, null, Lex.POSTGRESQL));
return calciteConnection;
}
}自定義模式
package com.example.multids.schema;
import org.apache.calcite.schema.impl.AbstractSchema;
import java.util.Map;
public class MySchemas extends AbstractSchema {
@Override
protected Map<String, org.apache.calcite.schema.Table> getTableMap() {
// 返回表映射，這里不需要額外處理
return super.getTableMap();
}
}查詢控制器
package com.example.multids.controller;
import com.example.multids.factory.DataSourceFactory;
import org.apache.calcite.jdbc.CalciteConnection;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import javax.sql.DataSource;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
@RestController
publicclass QueryController {
privatefinal DataSource mysqlDataSource;
privatefinal DataSource postgresDataSource;
@Autowired
public QueryController(@Qualifier("mysqlDataSource") DataSource mysqlDataSource,
@Qualifier("postgresDataSource") DataSource postgresDataSource) {
this.mysqlDataSource = mysqlDataSource;
this.postgresDataSource = postgresDataSource;
}
@GetMapping("/query")
public List<List<String>> query(@RequestParam String sql) throws SQLException {
// 創建Calcite連接
CalciteConnection connection = DataSourceFactory.createConnection(mysqlDataSource, postgresDataSource);
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
// 處理查詢結果
List<List<String>> result = new ArrayList<>();
while (resultSet.next()) {
int columnCount = resultSet.getMetaData().getColumnCount();
List<String> row = new ArrayList<>();
for (int i = 1; i <= columnCount; i++) {
row.add(resultSet.getString(i));
}
result.add(row);
}
// 關閉資源
resultSet.close();
statement.close();
connection.close();
return result;
}
}Application
package com.example.multids;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class MultidsApplication {
public static void main(String[] args) {
SpringApplication.run(MultidsApplication.class, args);
}
}測試
MySQL orders 表
CREATE TABLE orders (
id INT PRIMARY KEY,
user_id INT,
amount DECIMAL(10, 2),
order_date DATETIME
);PostgreSQL users 表
CREATE TABLE users (
id SERIAL PRIMARY KEY,
name VARCHAR(100),
email VARCHAR(100)
);測試執行一個聯合查詢，從兩個不同的數據源中獲取數據，SQL語句是:
SELECT o.id AS order_id, u.name AS user_name, o.amount, o.order_date
FROM orders o
JOIN users u ON o.user_id = u.id;測試結果
$ curl -X GET "http://localhost:8080/query?sql=SELECT%20o.id%20AS%20order_id,%20u.name%20AS%20user_name,%20o.amount,%20o.order_date%20FROM%20orders%20o%20JOIN%20users%20u%20ON%20o.user_id%20=%20u.id"
[
["1", "Alice", "199.99", "2025-04-10 21:30:00"],
["2", "Bob", "250.75", "2025-04-10 20:45:00"]
]責任編輯：武曉燕
來源：
Java知識日歷
