SpringBoot與Calcite整合，實現(xiàn)多數(shù)據(jù)源統(tǒng)一查詢系統(tǒng)
作者：Java知識日歷
最近，接到一個電商系統(tǒng)的兼職小單，其中訂單信息存儲在MySQL數(shù)據(jù)庫，而用戶信息存儲在PostgreSQL數(shù)據(jù)庫。客戶那邊想有一個統(tǒng)一查詢接口，可以通過SQL查詢同時獲取這兩個數(shù)據(jù)源的信息。
最近，接到一個電商系統(tǒng)的兼職小單，其中訂單信息存儲在MySQL數(shù)據(jù)庫，而用戶信息存儲在PostgreSQL數(shù)據(jù)庫。客戶那邊想有一個統(tǒng)一查詢接口，可以通過SQL查詢同時獲取這兩個數(shù)據(jù)源的信息。
為什么選擇Apache Calcite？
簡化開發(fā)流程
- 抽象層次高: Apache Calcite 提供了高層次的抽象，使得開發(fā)者可以專注于業(yè)務(wù)邏輯，而不必處理底層的數(shù)據(jù)庫連接和查詢執(zhí)行細(xì)節(jié)。
- 減少重復(fù)工作: 通過使用Calcite，可以避免重復(fù)造輪子，節(jié)省開發(fā)時間和成本。
強(qiáng)大的SQL解析和優(yōu)化能力
- SQL標(biāo)準(zhǔn)支持: Apache Calcite 支持多種SQL方言（如MySQL、PostgreSQL等），可以無縫地處理不同數(shù)據(jù)庫的SQL語句。
- 查詢優(yōu)化: 內(nèi)置的查詢優(yōu)化器可以根據(jù)不同的數(shù)據(jù)源特性進(jìn)行智能優(yōu)化，提高查詢性能。
靈活性和可擴(kuò)展性
- 自定義模式和表: 可以通過編程方式動態(tài)地添加和管理多個數(shù)據(jù)源，每個數(shù)據(jù)源可以有不同的模式和表結(jié)構(gòu)。
- 插件機(jī)制: 支持各種插件，可以根據(jù)需求靈活擴(kuò)展功能，例如自定義函數(shù)、聚合操作等。
高性能
- 內(nèi)存計算: Apache Calcite 支持內(nèi)存中的數(shù)據(jù)處理，減少了I/O開銷，提高了查詢速度。
- 分布式計算: 雖然本項目主要關(guān)注單機(jī)版實現(xiàn)，但Apache Calcite也可以擴(kuò)展到分布式環(huán)境中，支持大規(guī)模數(shù)據(jù)集的處理。
集成性強(qiáng)
- 與其他工具集成: 支持與其他大數(shù)據(jù)工具和技術(shù)棧（如Apache Flink、Presto等）集成，形成完整的數(shù)據(jù)分析解決方案。
哪些公司使用了Apache Calcite？
- Google 在其內(nèi)部的一些數(shù)據(jù)處理系統(tǒng)中使用 Apache Calcite，特別是在需要高性能和靈活性的場景下。
- IBM 在其數(shù)據(jù)倉庫和分析解決方案中使用 Apache Calcite，以提高查詢性能和靈活性。
- Intel 使用 Apache Calcite 來支持其大數(shù)據(jù)分析工具和解決方案，特別是在內(nèi)存計算方面。
- Alibaba Cloud: 阿里巴巴云在其大數(shù)據(jù)平臺中使用 Apache Calcite 提供強(qiáng)大的查詢優(yōu)化和執(zhí)行能力。
- MaxCompute (ODPS): 阿里巴巴的大規(guī)模數(shù)據(jù)計算服務(wù) MaxCompute 使用 Calcite 進(jìn)行 SQL 查詢處理。
- Elasticsearch 的某些高級功能，如 Kibana 中的復(fù)雜查詢，依賴于 Apache Calcite 進(jìn)行 SQL 解析和優(yōu)化。
- Netflix 使用 Apache Calcite 來構(gòu)建其內(nèi)部的數(shù)據(jù)虛擬化層，支持復(fù)雜的查詢和數(shù)據(jù)分析需求。
- Microsoft 在其一些大數(shù)據(jù)產(chǎn)品和服務(wù)中使用 Apache Calcite，例如 Azure Synapse Analytics。
- Teradata 使用 Apache Calcite 來增強(qiáng)其數(shù)據(jù)庫系統(tǒng)的查詢優(yōu)化和執(zhí)行性能。
- Uber 使用 Apache Calcite 來處理其龐大的數(shù)據(jù)集，并支持復(fù)雜的查詢和數(shù)據(jù)分析需求。
應(yīng)用場景
數(shù)據(jù)虛擬化
- 虛擬數(shù)據(jù)層: 創(chuàng)建一個虛擬的數(shù)據(jù)層，將分散在不同系統(tǒng)中的數(shù)據(jù)集中起來，提供統(tǒng)一的視圖。
- 動態(tài)數(shù)據(jù)源管理: 動態(tài)地添加和管理數(shù)據(jù)源，支持靈活的數(shù)據(jù)架構(gòu)設(shè)計。
商業(yè)智能 (BI) 工具
- 報表生成: 作為 BI 工具的核心組件，支持復(fù)雜的報表生成和數(shù)據(jù)分析。
- 自助服務(wù)分析: 提供自助服務(wù)分析功能，允許非技術(shù)人員進(jìn)行數(shù)據(jù)探索和分析。
機(jī)器學(xué)習(xí)與人工智能
- 特征工程: 在機(jī)器學(xué)習(xí)管道中使用 Calcite 進(jìn)行特征提取和數(shù)據(jù)準(zhǔn)備。
- 模型訓(xùn)練: 結(jié)合其他 AI 框架，利用 Calcite 進(jìn)行大規(guī)模數(shù)據(jù)集的查詢和處理。
多數(shù)據(jù)源查詢
- 統(tǒng)一接口訪問多個數(shù)據(jù)庫: 允許用戶通過單一接口查詢存儲在不同數(shù)據(jù)庫（如 MySQL、PostgreSQL、Oracle 等）中的數(shù)據(jù)。
- 聯(lián)合查詢: 支持跨數(shù)據(jù)源的復(fù)雜 SQL 查詢，例如從不同的數(shù)據(jù)庫中獲取相關(guān)聯(lián)的數(shù)據(jù)。
大數(shù)據(jù)平臺集成
- 與 Hadoop 生態(tài)系統(tǒng)集成: 與 Hive、HBase、Druid 等大數(shù)據(jù)工具結(jié)合，提供統(tǒng)一的查詢接口。
- 流處理與批處理: 支持 Apache Flink 和 Apache Beam 等流處理框架，實現(xiàn)實時數(shù)據(jù)分析。
嵌入式數(shù)據(jù)庫
- 輕量級數(shù)據(jù)庫引擎: 提供一個輕量級的 SQL 引擎，適用于嵌入式應(yīng)用程序和內(nèi)存數(shù)據(jù)庫。
- 內(nèi)存計算: 利用內(nèi)存計算加速查詢性能，適合需要快速響應(yīng)的應(yīng)用場景。
數(shù)據(jù)湖解決方案
- 統(tǒng)一元數(shù)據(jù)管理: 提供統(tǒng)一的元數(shù)據(jù)管理和查詢接口，方便數(shù)據(jù)湖的建設(shè)和維護(hù)。
- 多樣化數(shù)據(jù)格式支持: 支持多種數(shù)據(jù)格式（如 JSON、Parquet、ORC 等），滿足不同類型的數(shù)據(jù)存儲需求。
代碼實操
<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數(shù)據(jù)源配置
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數(shù)據(jù)源
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數(shù)據(jù)源
HikariConfig config = new HikariConfig();
config.setJdbcUrl("jdbc:postgresql://localhost:5432/user_db");
config.setUsername("postgres");
config.setPassword("postgres");
returnnew HikariDataSource(config);
}
}自定義數(shù)據(jù)源工廠
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" +
"}";
// 創(chuàng)建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 {
// 創(chuàng)建Calcite連接
CalciteConnection connection = DataSourceFactory.createConnection(mysqlDataSource, postgresDataSource);
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(sql);
// 處理查詢結(jié)果
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);
}
// 關(guān)閉資源
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)
);測試執(zhí)行一個聯(lián)合查詢，從兩個不同的數(shù)據(jù)源中獲取數(shù)據(jù)，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;測試結(jié)果
$ 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"]
]責(zé)任編輯：武曉燕
來源：
Java知識日歷
