JavaでInfluxDBを使用する

Posted on: 2019-10-18 2019-11-05
Tags: Persistence

1概要

InfluxDB

は時系列データ用の高性能ストアです** SQLライクなクエリ言語によるデータの挿入とリアルタイムクエリをサポートします。

この紹介記事では、InfluxDbサーバーへの接続方法、データベースの作成方法、時系列情報の作成方法、そしてデータベースへのクエリ方法について説明します。

2セットアップ

データベースに接続するには、

pom.xml

ファイルにエントリを追加する必要があります。

<dependency>
    <groupId>org.influxdb</groupId>
    <artifactId>influxdb-java</artifactId>
    <version>2.8</version>
</dependency>

この依存関係の最新版はhttps://search.maven.org/classic/#search%7C1%7Cg%3A%22org.influxdb%22%20AND%20a%3A%22influxdb-java%22にあります[Maven Central]

InfluxDBインスタンスも必要です。

データベースのダウンロードとインストール

の手順は、https://www.influxdata.com[InfluxData]Webサイトにあります。

3サーバーへの接続

3.1. 接続を作成する

データベース接続を作成するには、URL

String

とユーザー資格情報を接続ファクトリに渡す必要があります。

InfluxDB influxDB = InfluxDBFactory.connect(databaseURL, userName, password);

3.2. 接続を確認する

データベースとの通信はRESTful API ** を介して行われるため、永続的ではありません。

APIは、接続が機能していることを確認するための専用の「ping」サービスを提供します。接続が良好であれば、応答にデータベースのバージョンが含まれます。そうでなければ、それは__ “unknown”を含みます。

そのため、接続を作成した後、次のようにしてそれを確認できます。

Pong response = this.influxDB.ping();
if (response.getVersion().equalsIgnoreCase("unknown")) {
    log.error("Error pinging server.");
    return;
}

3.3. データベースを作成する

InfluxDBデータベースを作成することは、ほとんどのプラットフォームでデータベースを作成することに似ています。ただし、使用する前に少なくとも1つの保存ポリシーを作成する必要があります。

保存方針は、データを保存する期間をデータベースに指示します** CPUやメモリの統計などの時系列は、大きなデータセットに蓄積される傾向があります。

時系列データベースのサイズを制御するための典型的な戦略は

downsampling

です。「生の」データは高速で保存され、要約された後、短時間で削除されます。

保存ポリシーはデータを有効期限と関連付けることでこれを簡単にします** InfluxDataは彼らのサイトにhttps://docs.influxdata.com/influxdb/v1.4/guides/downsampling

and

retention/[詳細説明]を持っています。

データベースを作成したら、__defaultPolicyという名前の単一のポリシーを追加します。

influxDB.createDatabase("baeldung");
influxDB.createRetentionPolicy(
  "defaultPolicy", "baeldung", "30d", 1, true);

アイテム保持ポリシーを作成するには、

name、

database、

interval、

replication factor

（シングルインスタンスデータベースの場合は1）、およびそれがデフォルトポリシーであることを示す

boolean__が必要です。

3.4. ログレベルの設定

内部的には、InfluxDB APIは

Retrofit

を使用し、

logging interceptor。

を介してRetrofitのロギング機能へのインターフェースを公開します。

そのため、ロギングレベルを設定することができます。

influxDB.setLogLevel(InfluxDB.LogLevel.BASIC);

接続を開いてpingを実行するとメッセージが表示されます。

Dec 20, 2017 5:38:10 PM okhttp3.internal.platform.Platform log
INFO: --> GET http://127.0.0.1:8086/ping

使用可能なレベルは、

BASIC

、

FULL

、

HEADERS

、および__NONEです。

4データの追加と取得

4.1. ポイント

これで、データの挿入と取得を開始する準備が整いました。

InfluxDBの情報の基本的な単位は__Pointです。

メモリ使用率データを保持している点を見てみましょう。

Point point = Point.measurement("memory")
  .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
  .addField("name", "server1")
  .addField("free", 4743656L)
  .addField("used", 1015096L)
  .addField("buffer", 1010467L)
  .build();

メモリ統計として3つの

Longs

、ホスト名、およびタイムスタンプを含むエントリを作成しました。

これをデータベースに追加する方法を見てみましょう。

4.2. バッチを書く

時系列データは多くの小さな点からなる傾向があり、それらのレコードを一度に1つずつ書き込むことは非常に非効率的です。** 推奨される方法は、レコードをバッチにまとめることです。

InfluxDB APIは

BatchPoint

オブジェクトを提供します。

BatchPoints batchPoints = BatchPoints
  .database(dbName)
  .retentionPolicy("defaultPolicy")
  .build();

Point point1 = Point.measurement("memory")
  .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
  .addField("name", "server1")
  .addField("free", 4743656L)
  .addField("used", 1015096L)
  .addField("buffer", 1010467L)
  .build();

Point point2 = Point.measurement("memory")
  .time(System.currentTimeMillis() - 100, TimeUnit.MILLISECONDS)
  .addField("name", "server1")
  .addField("free", 4743696L)
  .addField("used", 1016096L)
  .addField("buffer", 1008467L)
  .build();

batchPoints.point(point1);
batchPoints.point(point2);
influxDB.write(batchPoints);

BatchPoint

を作成し、それに

Points

を追加します。タイムスタンプはプライマリインデックスなので、2番目のエントリのタイムスタンプを過去100ミリ秒に設定します。同じタイムスタンプで2つのポイントを送信した場合、1つだけが保持されます。

BatchPoints

をデータベースとアイテム保持ポリシーに関連付ける必要があります。

4.3. 一度に1つずつ書く

バッチ処理は、ユースケースによっては実用的でない場合があります。

InfluxDB接続を1回呼び出すだけでバッチモードを有効にしましょう。

influxDB.enableBatch(100, 200, TimeUnit.MILLISECONDS);

サーバーへの挿入または200ミリ秒ごとに送信するために100のバッチ処理を有効にしました。

バッチモードを有効にしても、1つずつ書き込むことができます。ただし、追加の設定が必要です。

influxDB.setRetentionPolicy("defaultPolicy");
influxDB.setDatabase(dbName);

さらに、今私たちは個々のポイントを書くことができます、そしてそれらはバックグラウンドスレッドによってバッチに集められています:

influxDB.write(point);

個々のポイントをエンキューする前に、（SQLの

use

コマンドに似た）** データベースを設定し、デフォルトの保存方針を設定する必要があります。行く道。
バッチモードは別のスレッドプールを利用します。そのため、不要になった場合は無効にすることをお勧めします。**

influxDB.disableBatch();

接続を閉じると、スレッドプールもシャットダウンされます。

influxDB.close();

4.4. クエリ結果のマッピング

クエリは

QueryResult

を返します。これはPOJOにマッピングできます。

クエリの構文を見る前に、メモリ統計を保持するクラスを作成しましょう。

@Measurement(name = "memory")
public class MemoryPoint {

    @Column(name = "time")
    private Instant time;

    @Column(name = "name")
    private String name;

    @Column(name = "free")
    private Long free;

    @Column(name = "used")
    private Long used;

    @Column(name = "buffer")
    private Long buffer;
}

このクラスには、

Points

の作成に使用した

Point.measurement（ “memory”）

に対応する

@ Measurement（name = “memory”）

という注釈が付けられています。

QueryResult

の各フィールドに、対応するフィールドの名前を含む

@ Column（name =“ XXX”）

注釈を追加します。

QueryResults

は、__InfluxDBResultMapperを使用してPOJOにマップされます。

4.5. InfluxDB

への問い合わせ

それでは、2ポイントバッチでデータベースに追加したポイントでPOJOを使用しましょう。

QueryResult queryResult = connection
  .performQuery("Select **  from memory", "baeldung");

InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
List<MemoryPoint> memoryPointList = resultMapper
  .toPOJO(queryResult, MemoryPoint.class);

assertEquals(2, memoryPointList.size());
assertTrue(4743696L == memoryPointList.get(0).getFree());

このクエリは、

memory

という名前の測定値が、選択可能な

Points

のテーブルとしてどのように格納されているかを示しています。

InfluxDBResultMapper

は、

QueryResult

を使用して

MemoryPoint.class

への参照を受け取り、ポイントのリストを返します。

結果をマッピングした後、クエリから受け取った

List

の長さを確認して、2つ受け取ったことを確認します。次に、リストの最初のエントリを見て、挿入した2番目のポイントの空きメモリサイズを確認します。 InfluxDBからのクエリ結果のデフォルトの順序はタイムスタンプの昇順です。

それを変更しましょう:

queryResult = connection.performQuery(
  "Select **  from memory order by time desc", "baeldung");
memoryPointList = resultMapper
  .toPOJO(queryResult, MemoryPoint.class);

assertEquals(2, memoryPointList.size());
assertTrue(4743656L == memoryPointList.get(0).getFree());

order by time desc__を追加すると、結果の順序が逆になります。

InfluxDBクエリはSQLとよく似ています。

彼らのサイトに

広範囲のリファレンスガイドがあります。

5結論

InfluxDBサーバーに接続し、アイテム保持ポリシーを使用してデータベースを作成し、サーバーにデータを挿入して取得しました。

例の完全なソースコードはhttps://github.com/eugenp/tutorials/tree/master/persistence-modules/influxdbです。

1概要