Java
- 写入数据
- 查询数据

Java

GreptimeDB 使用不同的客户端库来写入和查询数据。你根据需求可以选择合适的客户端库。

写入数据

GreptimeDB 提供了一个 ingester 库来帮助你写入数据。它使用 gRPC 协议，支持自动生成表结构，无需在写入数据前创建表。更多信息请参考自动生成表结构。

GreptimeDB 提供的 Java ingester SDK 是一个轻量级库，具有以下特点：

基于 SPI 的可扩展网络传输层，提供了使用 gRPC 框架的默认实现。
非阻塞、纯异步的易于使用的 API。
默认情况下自动收集各种性能指标，然后可以配置并将其写入本地文件。
能够对关键对象进行内存快照，配置并将其写入本地文件。这对于解决复杂问题很有帮助。

安装

安装 Java 开发工具包（JDK）

确保你的系统已安装 JDK 8 或更高版本。有关如何检查 Java 版本并安装 JDK 的更多信息，请参见 Oracle JDK 安装概述文档

将 GreptimeDB Java SDK 添加为依赖项

如果你使用的是 Maven，请将以下内容添加到 pom.xml 的依赖项列表中：

<dependency>
    <groupId>io.greptime</groupId>
    <artifactId>ingester-all</artifactId>
    <version>0.7.3</version>
</dependency>

最新版本可以在这里查看。

配置依赖项后，请确保它们对项目可用，这可能需要在 IDE 中刷新项目或运行依赖项管理器。

连接数据库

连接 GreptimeDB 时，通常需要用户名和密码。关于如何设置 GreptimeDB 的鉴权方式，请参考鉴权。这里我们在使用 ingester 库连接 GreptimeDB 时设置用户名和密码。

下方的代码展示了以最简单的配置连接到 GreptimeDB 的方法。如果想要自定义连接选项，请参考 API 文档。请注意每个选项的注释，它们提供了对其各自角色的详细解释。

// GreptimeDB 默认 database 为 "public"，默认 catalog 为 "greptime"，
// 我们可以将其作为测试数据库使用
String database = "public";
// 默认情况下，GreptimeDB 使用 gRPC 协议在监听端口 4001。
// 我们可以提供多个指向同一 GreptimeDB 集群的 endpoints，
// 客户端将根据负载均衡策略调用这些 endpoints。
String[] endpoints = {"127.0.0.1:4001"};
// 设置鉴权信息
AuthInfo authInfo = new AuthInfo("username", "password");
GreptimeOptions opts = GreptimeOptions.newBuilder(endpoints, database)
        // 如果数据库不需要鉴权，我们可以使用 AuthInfo.noAuthorization() 作为参数。
        .authInfo(authInfo)
        // 如果服务配置了 TLS ，设置 TLS 选项来启用安全连接
        //.tlsOptions(new TlsOptions())
        // 好的开始 ^_^
        .build();
GreptimeDB client = GreptimeDB.create(opts);

数据模型

表中的每条行数据包含三种类型的列：Tag、Timestamp 和 Field。更多信息请参考数据模型。列值的类型可以是 String、Float、Int、Timestamp 等。更多信息请参考数据类型。

低层级 API

GreptimeDB 的低层级 API 通过向具有预定义模式的 table 对象添加 row 来写入数据。

创建行数据

以下代码片段首先构建了一个名为 cpu_metric 的表，其中包括 host、cpu_user、cpu_sys 和 ts 列。随后，它向表中插入了一行数据。

该表包含三种类型的列：

Tag：host 列，值类型为 String。
Field：cpu_user 和 cpu_sys 列，值类型为 Float。
Timestamp：ts 列，值类型为 Timestamp。

// 为 CPU 指标构建表结构
TableSchema cpuMetricSchema = TableSchema.newBuilder("cpu_metric")
        .addTag("host", DataType.String) // 主机的标识符
        .addTimestamp("ts", DataType.TimestampMillisecond) // 毫秒级的时间戳
        .addField("cpu_user", DataType.Float64) // 用户进程的 CPU 使用率
        .addField("cpu_sys", DataType.Float64) // 系统进程的 CPU 使用率
        .build();
// 根据定义的模式创建表
Table cpuMetric = Table.from(cpuMetricSchema);
// 单行的示例数据
String host = "127.0.0.1"; // 主机标识符
long ts = System.currentTimeMillis(); // 当前时间戳
double cpuUser = 0.1; // 用户进程的 CPU 使用率（百分比）
double cpuSys = 0.12; // 系统进程的 CPU 使用率（百分比）
// 将一行数据插入表中
// 注意：参数必须按照定义的表结构的列顺序排列：host, ts, cpu_user, cpu_sys
cpuMetric.addRow(host, ts, cpuUser, cpuSys);

为了提高写入数据的效率，你可以一次创建多行数据以便写入到 GreptimeDB。

// 创建表结构
TableSchema cpuMetricSchema = TableSchema.newBuilder("cpu_metric")
        .addTag("host", DataType.String)
        .addTimestamp("ts", DataType.TimestampMillisecond)
        .addField("cpu_user", DataType.Float64)
        .addField("cpu_sys", DataType.Float64)
        .build();
TableSchema memMetricSchema = TableSchema.newBuilder("mem_metric")
        .addTag("host", DataType.String)
        .addTimestamp("ts", DataType.TimestampMillisecond)
        .addField("mem_usage", DataType.Float64)
        .build();
Table cpuMetric = Table.from(cpuMetricSchema);
Table memMetric = Table.from(memMetricSchema);
// 添加行数据
for (int i = 0; i < 10; i++) {
    String host = "127.0.0." + i;
    long ts = System.currentTimeMillis();
    double cpuUser = i + 0.1;
    double cpuSys = i + 0.12;
    cpuMetric.addRow(host, ts, cpuUser, cpuSys);
}
for (int i = 0; i < 10; i++) {
    String host = "127.0.0." + i;
    long ts = System.currentTimeMillis();
    double memUsage = i + 0.2;
    memMetric.addRow(host, ts, memUsage);
}

插入数据

下方示例展示了如何向 GreptimeDB 的表中插入行数据。

// 插入数据
// 考虑到尽可能提升性能和降低资源占用，SDK 设计为纯异步的。
// 返回值是一个 future 对象。如果你想立即获取结果，可以调用 `future.get()`。
CompletableFuture<Result<WriteOk, Err>> future = greptimeDB.write(cpuMetric, memMetric);
Result<WriteOk, Err> result = future.get();
if (result.isOk()) {
    LOG.info("Write result: {}", result.getOk());
} else {
    LOG.error("Failed to write: {}", result.getErr());
}

流式插入

当你需要插入大量数据时，例如导入历史数据，流式插入是非常有用的。

StreamWriter<Table, WriteOk> writer = greptimeDB.streamWriter();
// 写入数据到流中
writer.write(cpuMetric);
writer.write(memMetric);
// 你可以对流执行操作，例如删除前 5 行
writer.write(cpuMetric.subRange(0, 5), WriteOp.Delete);

在所有数据写入完毕后关闭流式写入。一般情况下，连续写入数据时不需要关闭流式写入。

// 完成流式写入
CompletableFuture<WriteOk> future = writer.completed();
WriteOk result = future.get();
LOG.info("Write result: {}", result);

更新数据

关于更新机制，请参考更新数据。下方代码首先保存了一行数据，然后使用相同的标签和时间索引来更新特定的行数据。

Table cpuMetric = Table.from(myMetricCpuSchema);
// 插入一行数据
long ts = 1703832681000L;
cpuMetric.addRow("host1", ts, 0.23, 0.12);
Result<WriteOk, Err> putResult = greptimeDB.write(cpuMetric).get();
// 更新行数据
Table newCpuMetric = Table.from(myMetricCpuSchema);
// 相同的标签 `host1`
// 相同的时间索引 `1703832681000`
// 新的值：cpu_user = `0.80`, cpu_sys = `0.11`
long ts = 1703832681000L;
myMetricCpuSchema.addRow("host1", ts, 0.80, 0.11);
// 覆盖现有数据
CompletableFuture<Result<WriteOk, Err>> future = greptimeDB.write(myMetricCpuSchema);
Result<WriteOk, Err> result = future.get();

高层级 API

SDK 的高层级 API 使用 ORM 风格的对象写入数据，它允许你以更面向对象的方式创建、插入和更新数据，为开发者提供了更友好的体验。然而，高层级 API 不如低层级 API 高效。这是因为 ORM 风格的对象在转换对象时可能会消耗更多的资源和时间。

创建行数据

GreptimeDB Java Ingester SDK 允许我们使用基本的 POJO 对象进行写入。虽然这种方法需要使用 Greptime 的注解，但它们很容易使用。

@Metric(name = "cpu_metric")
public class Cpu {
    @Column(name = "host", tag = true, dataType = DataType.String)
    private String host;
    @Column(name = "ts", timestamp = true, dataType = DataType.TimestampMillisecond)
    private long ts;
    @Column(name = "cpu_user", dataType = DataType.Float64)
    private double cpuUser;
    @Column(name = "cpu_sys", dataType = DataType.Float64)
    private double cpuSys;
    // getters and setters
    // ...
}
@Metric(name = "mem_metric")
public class Memory {
    @Column(name = "host", tag = true, dataType = DataType.String)
    private String host;
    @Column(name = "ts", timestamp = true, dataType = DataType.TimestampMillisecond)
    private long ts;
    @Column(name = "mem_usage", dataType = DataType.Float64)
    private double memUsage;
    // getters and setters
    // ...
}
// 添加行
List<Cpu> cpus = new ArrayList<>();
for (int i = 0; i < 10; i++) {
    Cpu c = new Cpu();
    c.setHost("127.0.0." + i);
    c.setTs(System.currentTimeMillis());
    c.setCpuUser(i + 0.1);
    c.setCpuSys(i + 0.12);
    cpus.add(c);
}
List<Memory> memories = new ArrayList<>();
for (int i = 0; i < 10; i++) {
    Memory m = new Memory();
    m.setHost("127.0.0." + i);
    m.setTs(System.currentTimeMillis());
    m.setMemUsage(i + 0.2);
    memories.add(m);
}

插入数据

写入 POJO 对象：

// 插入数据
CompletableFuture<Result<WriteOk, Err>> puts = greptimeDB.writePOJOs(cpus, memories);
Result<WriteOk, Err> result = puts.get();
if (result.isOk()) {
    LOG.info("Write result: {}", result.getOk());
} else {
    LOG.error("Failed to write: {}", result.getErr());
}

流式插入

当你需要插入大量数据时，例如导入历史数据，流式插入是非常有用的。

StreamWriter<List<?>, WriteOk> writer = greptimeDB.streamWriterPOJOs();
// 写入数据到流中
writer.write(cpus);
writer.write(memories);
// 你可以对流执行操作，例如删除前 5 行
writer.write(cpus.subList(0, 5), WriteOp.Delete);

在所有数据写入完毕后关闭流式写入。一般情况下，连续写入数据时不需要关闭流式写入。

// 完成流式写入
CompletableFuture<WriteOk> future = writer.completed();
WriteOk result = future.get();
LOG.info("Write result: {}", result);

更新数据

关于更新机制，请参考更新数据。下方代码首先保存了一行数据，然后使用相同的标签和时间索引来更新特定的行数据。

Cpu cpu = new Cpu();
cpu.setHost("host1");
cpu.setTs(1703832681000L);
cpu.setCpuUser(0.23);
cpu.setCpuSys(0.12);
// 插入一行数据
Result<WriteOk, Err> putResult = greptimeDB.writePOJOs(cpu).get();
// 更新该行数据
Cpu newCpu = new Cpu();
// 相同的标签 `host1`
newCpu.setHost("host1");
// 相同的时间索引 `1703832681000`
newCpu.setTs(1703832681000L);
// 新的值: cpu_user = `0.80`, cpu_sys = `0.11`
cpu.setCpuUser(0.80);
cpu.setCpuSys(0.11);
// 覆盖现有数据
Result<WriteOk, Err> updateResult = greptimeDB.writePOJOs(newCpu).get();

请参考此处获取更多代码示例。

调试日志

ingester SDK 提供了用于调试的指标和日志。请参考 Metrics & Display 和 Magic Tools 了解如何启用或禁用日志。

Ingester 库参考

API 文档

查询数据

GreptimeDB 使用 SQL 作为主要查询语言，兼容 MySQL 和 PostgreSQL。因此，我们推荐使用成熟的 SQL 驱动来查询数据。

安装

如果你使用的是 Maven，请将以下内容添加到 pom.xml 的依赖项列表中：

<!-- MySQL 依赖 -->
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.33</version>
</dependency>

连接数据库

下方的例子展示了如何连接到 GreptimeDB：

这里我们使用 MySQL 作为示例来演示如何连接到 GreptimeDB。


public static Connection getConnection() throws IOException, ClassNotFoundException, SQLException {
    Properties prop = new Properties();
    prop.load(QueryJDBC.class.getResourceAsStream("/db-connection.properties"));
    String dbName = (String) prop.get("db.database-driver");
    String dbConnUrl = (String) prop.get("db.url");
    String dbUserName = (String) prop.get("db.username");
    String dbPassword = (String) prop.get("db.password");
    Class.forName(dbName);
    Connection dbConn = DriverManager.getConnection(dbConnUrl, dbUserName, dbPassword);
    return Objects.requireNonNull(dbConn, "Failed to make connection!");
}

你需要一个 properties 文件来存储数据库连接信息，将其放在 Resources 目录中并命名为 db-connection.properties。文件内容如下：

# DataSource
db.database-driver=com.mysql.cj.jdbc.Driver
db.url=jdbc:mysql://localhost:4002/public
db.username=
db.password=

或者你可以从这里获取文件。

Raw SQL

我们推荐使用 Raw SQL 来体验 GreptimeDB 的全部功能。下面的例子展示了如何使用 Raw SQL 查询数据：

try (Connection conn = getConnection()) {
    Statement statement = conn.createStatement();
    // DESC table;
    ResultSet rs = statement.executeQuery("DESC cpu_metric");
    LOG.info("Column | Type | Key | Null | Default | Semantic Type ");
    while (rs.next()) {
        LOG.info("{} | {} | {} | {} | {} | {}",
                rs.getString(1),
                rs.getString(2),
                rs.getString(3),
                rs.getString(4),
                rs.getString(5),
                rs.getString(6));
    }
    // SELECT COUNT(*) FROM cpu_metric;
    rs = statement.executeQuery("SELECT COUNT(*) FROM cpu_metric");
    while (rs.next()) {
        LOG.info("Count: {}", rs.getInt(1));
    }
    // SELECT * FROM cpu_metric ORDER BY ts DESC LIMIT 5;
    rs = statement.executeQuery("SELECT * FROM cpu_metric ORDER BY ts DESC LIMIT 5");
    LOG.info("host | ts | cpu_user | cpu_sys");
    while (rs.next()) {
        LOG.info("{} | {} | {} | {}",
                rs.getString("host"),
                rs.getTimestamp("ts"),
                rs.getDouble("cpu_user"),
                rs.getDouble("cpu_sys"));
    }
}

请参考此处获取更多可执行代码。

查询库参考

有关如何使用查询库的更多信息，请参考相应库的文档：

JDBC 在线教程

Java

Java

写入数据

安装

连接数据库

数据模型

低层级 API

创建行数据

插入数据

流式插入

更新数据

高层级 API

创建行数据

插入数据

流式插入

更新数据

更多示例

调试日志

Ingester 库参考

查询数据

推荐的查询库

安装

连接数据库

Raw SQL

查询库参考