快捷導(dǎo)航

Calcite使用SQL實(shí)現(xiàn)查詢csv/json文件內(nèi)容

更新時(shí)間：2025年01月22日 09:28:40 作者：張鐵牛

calcite可以支持文件系統(tǒng)的數(shù)據(jù)源適配, 其實(shí)官方已經(jīng)提供了相應(yīng)的能力, 其支持csv和json的查詢適配,下面我們就來看看Calcite如何使用SQL實(shí)現(xiàn)查詢csv/json文件內(nèi)容吧

1. 簡介

我們在前面的文章提到了calcite可以支持文件系統(tǒng)的數(shù)據(jù)源適配, 其實(shí)官方已經(jīng)提供了相應(yīng)的能力, 其支持csv和json的查詢適配, 廢話不多說, 直接展示.

2. Maven

<!-- calcite文件系統(tǒng)支持 -->
<dependency>
    <groupId>org.apache.calcite</groupId>
    <artifactId>calcite-file</artifactId>
    <version>1.37.0</version>
</dependency>
<dependency>
    <groupId>org.apache.calcite</groupId>
    <artifactId>calcite-core</artifactId>
    <version>1.37.0</version>
</dependency>

3. 數(shù)據(jù)文件準(zhǔn)備

3.1 csv

user_info.csv

首行將來被解析成表的字段, 冒號后面是字段類型, 如果未指定類型默認(rèn)使用varchar

ID:long,姓名:string,GENDER:string,BIRTHDAY:date
100,"張三",,"2001-01-01"
110,"李四","M","2001-01-01"
120,"王五","M","2002-05-03"
130,"趙六","F","2005-09-07"
140,"張鐵牛","M","2007-01-01"

3.2 json

role_info.json

[
  {
    "id": 123,
    "name": "管理員",
    "key": "manager"
  },
  {
    "id": 234,
    "name": "老師",
    "key": "teacher"
  },
  {
    "id": 345,
    "name": "學(xué)生",
    "key": "student"
  }
]

然后將文件放到resources/file目錄下

4. 核心代碼

package com.ldx.calcite;

import com.google.common.collect.ImmutableMap;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.calcite.adapter.file.FileSchemaFactory;
import org.apache.calcite.jdbc.CalciteConnection;
import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.util.Sources;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.testng.collections.Maps;

import java.net.URL;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Map;
import java.util.Properties;

@Slf4j
public class CalciteFileTest {
    private static Connection connection;

    private static SchemaPlus rootSchema;

    private static Statement statement;

    @BeforeAll
    @SneakyThrows
    public static void beforeAll() {
        Properties info = new Properties();
        // 不區(qū)分sql大小寫
        info.setProperty("caseSensitive", "false");
        // 創(chuàng)建Calcite連接
        connection = DriverManager.getConnection("jdbc:calcite:", info);
        CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class);
        // 構(gòu)建RootSchema，在Calcite中，RootSchema是所有數(shù)據(jù)源schema的parent，多個(gè)不同數(shù)據(jù)源schema可以掛在同一個(gè)RootSchema下
        rootSchema = calciteConnection.getRootSchema();
        final Schema schema = FileSchemaFactory.INSTANCE.create(rootSchema, "x",
                ImmutableMap.of("directory", resourcePath("file"), "flavor", "scannable"));
        rootSchema.add("test", schema);
        // 創(chuàng)建SQL語句執(zhí)行查詢
        statement = calciteConnection.createStatement();
    }

    @Test
    @SneakyThrows
    public void execute_simple_query() {
        ResultSet resultSet = statement.executeQuery("SELECT * FROM test.user_info");
        printResultSet(resultSet);
    }
  
    @Test
    @SneakyThrows
    public void test_execute_join_query() {
        ResultSet resultSet = statement.executeQuery("SELECT * FROM test.user_info ui inner join test.role_info ri on ui.role_id = ri.id");
        printResultSet(resultSet);
    }

    @AfterAll
    @SneakyThrows
    public static void closeResource() {
        statement.close();
        connection.close();
    }

    private static String resourcePath(String path) {
        final URL url = CalciteFileTest.class.getResource("/" + path);
        return Sources.of(url).file().getAbsolutePath();
    }

    public static void printResultSet(ResultSet resultSet) throws SQLException {
        // 獲取 ResultSet 元數(shù)據(jù)
        ResultSetMetaData metaData = resultSet.getMetaData();

        // 獲取列數(shù)
        int columnCount = metaData.getColumnCount();
        log.info("Number of columns: {}",columnCount);

        // 遍歷 ResultSet 并打印結(jié)果
        while (resultSet.next()) {
            final Map<String, String> item = Maps.newHashMap();
            // 遍歷每一列并打印
            for (int i = 1; i <= columnCount; i++) {
                String columnName = metaData.getColumnName(i);
                String columnValue = resultSet.getString(i);
                item.put(columnName, columnValue);
            }

            log.info(item.toString());
        }
    }
}

其實(shí)核心代碼就幾行, 如下:

通過FileSchemaFactory指定文件目錄和文件內(nèi)容的讀取方式, 默認(rèn)將指定目錄下的csv和json文件讀取成Table, 表名就是file的名稱

flavor:

SCANNABLE: 數(shù)據(jù)掃描。會(huì)更側(cè)重于快速地讀取和遍歷數(shù)據(jù)。這種方式適用于需要對大量數(shù)據(jù)進(jìn)行全表掃描或者范圍掃描的情況，例如統(tǒng)計(jì)匯總操作
FILTERABLE: 數(shù)據(jù)過濾。會(huì)更側(cè)重于數(shù)據(jù)的條件篩選，比如在 SQL 查詢中的WHERE子句。
TRANSLATABLE: 數(shù)據(jù)轉(zhuǎn)換。會(huì)更側(cè)重于數(shù)據(jù)轉(zhuǎn)換，以滿足特定的查詢需求或者數(shù)據(jù)處理要求。這種轉(zhuǎn)換可能包括數(shù)據(jù)類型的轉(zhuǎn)換（如將字符串類型的數(shù)字轉(zhuǎn)換為實(shí)際的數(shù)值類型）、格式轉(zhuǎn)換（如日期格式的調(diào)整）等。

// 這里的第二個(gè)參數(shù)“x”沒什么意義, 源碼中沒用到, 可以隨便填
final Schema schema = FileSchemaFactory.INSTANCE.create(rootSchema, "x",
               ImmutableMap.of("directory", resourcePath("file"), "flavor", "scannable"));
// 使用目錄名稱為schema名稱, 這里的test就是schema名稱
rootSchema.add("test", schema);

calcite也可以做對應(yīng)表的關(guān)聯(lián)查詢, 測試中csv關(guān)聯(lián)json文件信息

"SELECT * FROM test.user_info ui inner join test.role_info ri on ui.role_id = ri.id"