SpringBoot與Calcite整合，實現(xiàn)多數(shù)據(jù)源統(tǒng)一查詢系統(tǒng)

作者：Java知識日歷 2025-04-14 01:00:00

最近，接到一個電商系統(tǒng)的兼職小單，其中訂單信息存儲在MySQL數(shù)據(jù)庫，而用戶信息存儲在PostgreSQL數(shù)據(jù)庫。客戶那邊想有一個統(tǒng)一查詢接口，可以通過SQL查詢同時獲取這兩個數(shù)據(jù)源的信息。

為什么選擇Apache Calcite？

簡化開發(fā)流程

抽象層次高: Apache Calcite 提供了高層次的抽象，使得開發(fā)者可以專注于業(yè)務邏輯，而不必處理底層的數(shù)據(jù)庫連接和查詢執(zhí)行細節(jié)。
減少重復工作: 通過使用Calcite，可以避免重復造輪子，節(jié)省開發(fā)時間和成本。

強大的SQL解析和優(yōu)化能力

SQL標準支持: Apache Calcite 支持多種SQL方言（如MySQL、PostgreSQL等），可以無縫地處理不同數(shù)據(jù)庫的SQL語句。
查詢優(yōu)化: 內置的查詢優(yōu)化器可以根據(jù)不同的數(shù)據(jù)源特性進行智能優(yōu)化，提高查詢性能。

靈活性和可擴展性

自定義模式和表: 可以通過編程方式動態(tài)地添加和管理多個數(shù)據(jù)源，每個數(shù)據(jù)源可以有不同的模式和表結構。
插件機制: 支持各種插件，可以根據(jù)需求靈活擴展功能，例如自定義函數(shù)、聚合操作等。

高性能

內存計算: Apache Calcite 支持內存中的數(shù)據(jù)處理，減少了I/O開銷，提高了查詢速度。
分布式計算: 雖然本項目主要關注單機版實現(xiàn)，但Apache Calcite也可以擴展到分布式環(huán)境中，支持大規(guī)模數(shù)據(jù)集的處理。

集成性強

與其他工具集成: 支持與其他大數(shù)據(jù)工具和技術棧（如Apache Flink、Presto等）集成，形成完整的數(shù)據(jù)分析解決方案。

哪些公司使用了Apache Calcite？

Google 在其內部的一些數(shù)據(jù)處理系統(tǒng)中使用 Apache Calcite，特別是在需要高性能和靈活性的場景下。
IBM 在其數(shù)據(jù)倉庫和分析解決方案中使用 Apache Calcite，以提高查詢性能和靈活性。
Intel 使用 Apache Calcite 來支持其大數(shù)據(jù)分析工具和解決方案，特別是在內存計算方面。
Alibaba Cloud: 阿里巴巴云在其大數(shù)據(jù)平臺中使用 Apache Calcite 提供強大的查詢優(yōu)化和執(zhí)行能力。
MaxCompute (ODPS): 阿里巴巴的大規(guī)模數(shù)據(jù)計算服務 MaxCompute 使用 Calcite 進行 SQL 查詢處理。
Elasticsearch 的某些高級功能，如 Kibana 中的復雜查詢，依賴于 Apache Calcite 進行 SQL 解析和優(yōu)化。
Netflix 使用 Apache Calcite 來構建其內部的數(shù)據(jù)虛擬化層，支持復雜的查詢和數(shù)據(jù)分析需求。
Microsoft 在其一些大數(shù)據(jù)產(chǎn)品和服務中使用 Apache Calcite，例如 Azure Synapse Analytics。
Teradata 使用 Apache Calcite 來增強其數(shù)據(jù)庫系統(tǒng)的查詢優(yōu)化和執(zhí)行性能。
Uber 使用 Apache Calcite 來處理其龐大的數(shù)據(jù)集，并支持復雜的查詢和數(shù)據(jù)分析需求。

應用場景

數(shù)據(jù)虛擬化

虛擬數(shù)據(jù)層: 創(chuàng)建一個虛擬的數(shù)據(jù)層，將分散在不同系統(tǒng)中的數(shù)據(jù)集中起來，提供統(tǒng)一的視圖。
動態(tài)數(shù)據(jù)源管理: 動態(tài)地添加和管理數(shù)據(jù)源，支持靈活的數(shù)據(jù)架構設計。

商業(yè)智能 (BI) 工具

報表生成: 作為 BI 工具的核心組件，支持復雜的報表生成和數(shù)據(jù)分析。
自助服務分析: 提供自助服務分析功能，允許非技術人員進行數(shù)據(jù)探索和分析。

機器學習與人工智能

特征工程: 在機器學習管道中使用 Calcite 進行特征提取和數(shù)據(jù)準備。
模型訓練: 結合其他 AI 框架，利用 Calcite 進行大規(guī)模數(shù)據(jù)集的查詢和處理。

多數(shù)據(jù)源查詢

統(tǒng)一接口訪問多個數(shù)據(jù)庫: 允許用戶通過單一接口查詢存儲在不同數(shù)據(jù)庫（如 MySQL、PostgreSQL、Oracle 等）中的數(shù)據(jù)。
聯(lián)合查詢: 支持跨數(shù)據(jù)源的復雜 SQL 查詢，例如從不同的數(shù)據(jù)庫中獲取相關聯(lián)的數(shù)據(jù)。

大數(shù)據(jù)平臺集成

與 Hadoop 生態(tài)系統(tǒng)集成: 與 Hive、HBase、Druid 等大數(shù)據(jù)工具結合，提供統(tǒng)一的查詢接口。
流處理與批處理: 支持 Apache Flink 和 Apache Beam 等流處理框架，實現(xiàn)實時數(shù)據(jù)分析。

嵌入式數(shù)據(jù)庫

輕量級數(shù)據(jù)庫引擎: 提供一個輕量級的 SQL 引擎，適用于嵌入式應用程序和內存數(shù)據(jù)庫。
內存計算: 利用內存計算加速查詢性能，適合需要快速響應的應用場景。

數(shù)據(jù)湖解決方案

統(tǒng)一元數(shù)據(jù)管理: 提供統(tǒng)一的元數(shù)據(jù)管理和查詢接口，方便數(shù)據(jù)湖的建設和維護。
多樣化數(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);

        // 處理查詢結果
        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)
);

測試執(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;

測試結果

$ 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知識日歷

Calcite 電商系統(tǒng)MySQL