Java

Java

GreptimeDB 提供了用于高吞吐量数据写入的 ingester 库。它使用 gRPC 协议，支持自动生成表结构，无需在写入数据前创建表。更多信息请参考自动生成表结构。

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

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

快速开始 Demo

你可以通过快速开始 Demo 来了解如何使用 GreptimeDB Java SDK。

安装

安装 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 时设置了 --user-provider，则需要提供用户名和密码才能连接到 GreptimeDB。以下示例显示了使用 SDK 连接到 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);

高层级 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);

调试日志

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

Ingester 库参考

API 文档