Write multi-field data synchronously and asynchronously with the Java SDK - Time Series Database

The Lindorm SDK's multi-value model lets you write multiple fields per data point in a single call, similar to InfluxDB's multi-field measurement model. This document shows how to write data using the Aliyun TSDB SDK for Java, covering both synchronous and asynchronous writes.

Full examples are in the SDK repository under test/example/ExampleOfMultiField and test/com.aliyun.hitsdb.client/TestHiTSDBClientMultiFieldFeatures.

The multi-value model requires Lindorm SDK V0.2.0 or later. Asynchronous multi-value writes require V0.2.1 or later. SDKs earlier than V0.2.0 use a simulated multi-value model. We recommend that you do not use the simulated multi-value model. Data written with the MultiField class (V0.2.0 or later) cannot be queried using the MultiValue class of an earlier SDK version. For the latest SDK version and download link, see the "Versions" section in the SDK documentation.

Prerequisites

Before you begin, ensure that you have:

The Aliyun TSDB SDK for Java V0.2.0 or later (V0.2.1 or later for asynchronous writes)
A running Lindorm Time Series Database instance

Data model

Each MultiFieldPoint you write has four components:

Component	Description	Example
Metric	The top-level category, similar to a measurement in InfluxDB	`wind`
Fields	Sub-categories within a metric; a single metric can carry multiple fields	`speed`, `direction`, `description`
Tags	Metadata that identify a timeline	`sensor="95D8-7913"`, `city="hangzhou"`
Timestamp	Unix timestamp of the data point	`1537170208L`

Write data synchronously

Use multiFieldPutSync() for single-point or multi-point synchronous writes.

/*
 * Build a MultiFieldPoint using the builder API.
 * Metric:    top-level category (e.g., "wind")
 * Fields:    measurements within the metric (numeric or string values)
 * Tags:      timeline identifiers
 * Timestamp: Unix timestamp of the data point
 */
MultiFieldPoint multiFieldPoint = MultiFieldPoint.metric("wind")
        .field("speed", 45.2)
        .field("level", 1.2)
        .field("direction", "S")
        .field("description", "Breeze")
        .tag("sensor", "95D8-7913")
        .tag("city", "hangzhou")
        .tag("province", "zhejiang")
        .timestamp(1537170208L)
        .build();

// Synchronous write
tsdb.multiFieldPutSync(multiFieldPoint);

Write data asynchronously

Asynchronous writes use multiFieldPut() with a buffered queue. Configure the queue size, consumer threads, and an optional callback via TSDBConfig.

Choose a callback type

Callback type	Class	`response()` returns	`failed()` returns
None	(no callback)	—	—
Common	`MultiFieldBatchPutCallback`	Points that succeeded, plus the result	Points that failed, plus the exception
Summary	`MultiFieldBatchPutSummaryCallback`	Points processed, plus `SummaryResult` (success/failure counts)	Points that failed, plus the exception
Detail	`MultiFieldBatchPutDetailsCallback`	Points processed, plus `DetailsResult` (failed points and their errors)	Points that failed, plus the exception

Use the common callback for basic success/failure handling. Use the summary callback when you need aggregate counts. Use the detail callback when you need per-point error details.

Shared configuration

All async write examples use the same base TSDBConfig. Configure it once per application and reuse it:

TSDBConfig config = TSDBConfig.address("127.0.0.1", 8242)
        .httpConnectTimeout(90)
        // Batch size for single-value async writes
        .batchPutSize(500)
        // Consumer threads for single-value async writes
        .batchPutConsumerThreadCount(1)
        // Buffer queue size for multi-value async writes
        .multiFieldBatchPutBufferSize(10000)
        // Consumer threads for multi-value async writes
        .multiFieldBatchPutConsumerThreadCount(1)
        // Attach a callback here (see options below)
        .listenMultiFieldBatchPut(callback)
        .config();

// Create the TSDB instance once per application lifecycle
TSDB tsdb = TSDBClientFactory.connect(config);

Important

Create the TSDB instance only once per application. Do not close it until the application exits.

No callback

Omit .listenMultiFieldBatchPut() from the config. The client submits writes without returning results.

TSDBConfig config = TSDBConfig.address("127.0.0.1", 8242)
        .batchPutSize(500)
        .batchPutConsumerThreadCount(1)
        .multiFieldBatchPutBufferSize(10000)
        .multiFieldBatchPutConsumerThreadCount(1)
        .config();

TSDB tsdb = TSDBClientFactory.connect(config);

MultiFieldPoint point = MultiFieldPoint
        .metric("test-test-test")
        .tag("a", "1")
        .tag("b", "2")
        .timestamp(System.currentTimeMillis())
        .field("f1", Math.random())
        .field("f2", Math.random())
        .build();
tsdb.multiFieldPut(point);

// Close only when the application exits
tsdb.close();

Common callback

MultiFieldBatchPutCallback reports successes and failures as they occur.

AbstractMultiFieldBatchPutCallback callback = new MultiFieldBatchPutCallback() {

    final AtomicInteger num = new AtomicInteger();

    @Override
    public void response(String address, List<MultiFieldPoint> points, Result result) {
        int count = num.addAndGet(points.size());
        System.out.println(result);
        System.out.println("The system has processed" + count + "points.");
    }

    @Override
    public void failed(String address, List<MultiFieldPoint> points, Exception ex) {
        System.err.println("The callback function fails." + points.size() + " error!");
        ex.printStackTrace();
    }
};

TSDBConfig config = TSDBConfig.address("127.0.0.1", 8242)
        .httpConnectTimeout(90)
        .batchPutSize(500)
        .batchPutConsumerThreadCount(1)
        .multiFieldBatchPutBufferSize(10000)
        .multiFieldBatchPutConsumerThreadCount(1)
        .listenMultiFieldBatchPut(callback)
        .config();

TSDB tsdb = TSDBClientFactory.connect(config);

MultiFieldPoint point = MultiFieldPoint
        .metric("test-test-test")
        .tag("a", "1")
        .tag("b", "2")
        .timestamp(System.currentTimeMillis())
        .field("f1", Math.random())
        .field("f2", Math.random())
        .build();
tsdb.multiFieldPut(point);

tsdb.close();

Summary callback

MultiFieldBatchPutSummaryCallback returns aggregate success and failure counts in SummaryResult.

AbstractMultiFieldBatchPutCallback callback = new MultiFieldBatchPutSummaryCallback() {

    final AtomicInteger num = new AtomicInteger();

    @Override
    public void response(String address, List<MultiFieldPoint> points, SummaryResult result) {
        int count = num.addAndGet(points.size());
        System.out.println(result);
        System.out.println("The system has processed" + count + "points.");
    }

    @Override
    public void failed(String address, List<MultiFieldPoint> points, Exception ex) {
        System.err.println("The callback function fails" + points.size() + " error!");
        ex.printStackTrace();
    }
};

TSDBConfig config = TSDBConfig.address("127.0.0.1", 8242)
        .httpConnectTimeout(90)
        .batchPutSize(500)
        .batchPutConsumerThreadCount(1)
        .multiFieldBatchPutBufferSize(10000)
        .multiFieldBatchPutConsumerThreadCount(1)
        .listenMultiFieldBatchPut(callback)
        .config();

TSDB tsdb = TSDBClientFactory.connect(config);

MultiFieldPoint point = MultiFieldPoint
        .metric("test-test-test")
        .tag("a", "1")
        .tag("b", "2")
        .timestamp(System.currentTimeMillis())
        .field("f1", Math.random())
        .field("f2", Math.random())
        .build();
tsdb.multiFieldPut(point);

tsdb.close();

Detail callback

MultiFieldBatchPutDetailsCallback returns per-point error details in DetailsResult.

MultiFieldBatchPutDetailsCallback callback = new MultiFieldBatchPutDetailsCallback() {

    final AtomicInteger num = new AtomicInteger();

    @Override
    public void response(String address, List<MultiFieldPoint> points, DetailsResult result) {
        int count = num.addAndGet(points.size());
        System.out.println(result);
        System.out.println("The system has processed" + count + "points.");
    }

    @Override
    public void failed(String address, List<MultiFieldPoint> points, Exception ex) {
        System.err.println("The callback function fails." + points.size() + " error!");
        ex.printStackTrace();
    }
};

TSDBConfig config = TSDBConfig.address("127.0.0.1", 8242)
        .httpConnectTimeout(90)
        .batchPutSize(500)
        .batchPutConsumerThreadCount(1)
        .multiFieldBatchPutBufferSize(10000)
        .multiFieldBatchPutConsumerThreadCount(1)
        .listenMultiFieldBatchPut(callback)
        .config();

TSDB tsdb = TSDBClientFactory.connect(config);

MultiFieldPoint point = MultiFieldPoint
        .metric("test-test-test")
        .tag("a", "1")
        .tag("b", "2")
        .timestamp(System.currentTimeMillis())
        .field("f1", Math.random())
        .field("f2", Math.random())
        .build();
tsdb.multiFieldPut(point);

tsdb.close();