[SpringBoot] InfluxDB 연동 및 예제

시계열 데이터베이스 InfluxDB에 대한 설명과 설치는 아래 포스팅을 참고해주세요.

[TSDB/InfluxDB] 시계열 데이터베이스 InfluxDB 설치하기

 

 

우선 influxDB의 UI에 접근합니다. (http://localhost:8086/)

메인 화면에 Java가 없다면 [MORE →] 버튼을 클릭합니다.

 

java를 검색한 후 클릭합니다.

 

그러면 JAVA를 사용할 경우 추가해야 할 Dependency라던가 Client 설정, 데이터 저장 등의 기본적인 가이드 코드를 제공하는 것을 확인할 수 있습니다.

Add Dependency

<dependency>
  <groupId>com.influxdb</groupId>
  <artifactId>influxdb-client-java</artifactId>
  <version>6.3.0</version>
</dependency>

 

Initialize the Client

package example;

import java.time.Instant;
import java.util.List;

import com.influxdb.annotations.Column;
import com.influxdb.annotations.Measurement;
import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
import com.influxdb.client.WriteApi;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import com.influxdb.query.FluxTable;

public class InfluxDB2Example {
  public static void main(final String[] args) {

    // You can generate an API token from the "API Tokens Tab" in the UI
    String token = System.getenv("INFLUX_TOKEN");
    String bucket = "launcher";
    String org = "admin";

    InfluxDBClient client = InfluxDBClientFactory.create("http://localhost:18086", token.toCharArray());
  }
}

 

Write Data

Option 1: Use InfluxDB Line Protocol to write data

String data = "mem,host=host1 used_percent=23.43234543";

WriteApiBlocking writeApi = client.getWriteApiBlocking();
writeApi.writeRecord(bucket, org, WritePrecision.NS, data);

 Option 2: Use a Data Point to write data

Point point = Point
  .measurement("mem")
  .addTag("host", "host1")
  .addField("used_percent", 23.43234543)
  .time(Instant.now(), WritePrecision.NS);

WriteApiBlocking writeApi = client.getWriteApiBlocking();
writeApi.writePoint(bucket, org, point);

 Option 3: Use POJO and corresponding class to write data

Mem mem = new Mem();
mem.host = "host1";
mem.used_percent = 23.43234543;
mem.time = Instant.now();

WriteApiBlocking writeApi = client.getWriteApiBlocking();
writeApi.writeRecord(bucket, org, WritePrecision.NS, mem);

@Measurement(name = "mem")
public static class Mem {
  @Column(tag = true)
  String host;
  @Column
  Double used_percent;
  @Column(timestamp = true)
  Instant time;
}

 

Execute a Flux query

String query = "from(bucket: \"launcher\") |> range(start: -1h)";
List<FluxTable> tables = client.getQueryApi().query(query, org);

for (FluxTable table : tables) {
  for (FluxRecord record : table.getRecords()) {
    System.out.println(record);
  }
}

 

Dispose the Client

client.close();

 

가이드대로 먼저 Maven Dependency를 추가합니다.

		<dependency>
			<groupId>com.influxdb</groupId>
			<artifactId>influxdb-client-java</artifactId>
			<version>3.3.0</version>
		</dependency>

 

influxDB Client 구성을 위한 속성을 application.yml에 추가합니다.

spring:
  influx:
    url: http://localhost:8086
    user: {Username}
    password: {Password}
    token: {API Token}
    org: {Organization Name}
    bucket: {Bucket Name}

user, password, org, bucket은 influxDB의 UI에 처음 접근했을 때 입력했던 정보를 넣으면 되고,

token 발급은 influxdata 문서를 참고하면 되는데, 버전에 따라 조금 다르지만 대개 Load Data > API TOKENS에서 발급받을 수 있습니다.

 

발급받은 토큰까지 설정 완료했다면, InfluxDBConfiguration.java를 생성하여 influxDBClient를 구성합니다.

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class InfluxDBConfiguration {

	@Value("${spring.influx.url}")
	private String url;

	@Value("${spring.influx.token}")
	private char[] token;

	@Value("${spring.influx.org}")
	private String org;

	@Value("${spring.influx.bucket}")
	private String bucket;

	@Bean
	public InfluxDBClient influxDBClient() {
		return InfluxDBClientFactory.create(url, token, org, bucket);
	}
}

 

influxDBClient로 데이터를 INSERT 할 수 있습니다.

    @Resource
    private InfluxDBClient influxDBClient;
    
    public void addTSData() {
        Point row = Point.measurement("launcher_client_connection")
                         .addTag("privateIp", vo.getPrivateIp())
                         .addTag("port", vo.getPort())
                         .addField("clientCnt", vo.getClientCnt());
        influxDBClient.getWriteApiBlocking().writePoint(row);
    }

 

 

InfluxDB UI에 접속하여 INSERT 한 데이터로 그래프를 그려보겠습니다.

influxDB v2부터 influxql이 없어지면서 flux query를 새로 학습해야 하는 번거로움이 있는데, Data Explorer 하단에는 UI로 손쉽게 데이터를 설정할 수 있습니다.

[SCRIPT EDITOR] 버튼을 누르면 아래와 같이 스크립트로 작성 가능합니다.

from(bucket: "launcher")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "launcher_client_connection")
  |> filter(fn: (r) => r["_field"] == "clientCnt")
  |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
  |> yield(name: "launcher_client")

from(bucket: "launcher")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r["_measurement"] == "launcher_client_connection")
  |> group(columns: ["_measurement"], mode:"by")
  |> aggregateWindow(every: v.windowPeriod, fn: sum, createEmpty: false)
  |> yield(name: "total")

특정 기간 조회

import "experimental"

from(bucket: "launcher")
  |> range(start: experimental.subDuration(from: now(), d: 1w), stop: now())
  |> filter(fn: (r) => r["_measurement"] == "launcher_client_connection")
  |> aggregateWindow(every: 10m, fn: mean, createEmpty: false)
  |> yield(name: "launcher_client")

import "experimental"

from(bucket: "launcher")
  |> range(start: experimental.subDuration(from: now(), d: 1w), stop: now())
  |> filter(fn: (r) => r["_measurement"] == "launcher_client_connection")
  |> group(columns: ["_measurement"], mode:"by")
  |> aggregateWindow(every: 10m, fn: sum, createEmpty: false)
  |> map(fn: (r) => ({r with _value: r._value / 60 }))
  |> yield(name: "total")

 

[SUBMIT] 버튼을 누르면 차트가 그려지고, Dashboard에 저장하여 손쉽게 실시간 데이터를 확인할 수 있습니다.

 

이전에 번거롭게 UI를 그렸던 포스팅과 비교하면 비교적 간편하게 여러 차트 UI로 데이터를 확인할 수 있습니다.

 

 

Data Explorer

 

Dashboard

 

Notebook