Lindorm:Flink を使用して LindormTSDB にデータを書き込む

前提条件

開始する前に、以下の条件を満たしていることを確認してください。

• Realtime Compute for Apache Flink を有効化するか、自社管理型の Flink サービスをセットアップします。詳細については、「Realtime Compute for Apache Flink の有効化」をご参照ください。

  説明

  Realtime Compute for Apache Flink で使用される Ververica Runtime (VVR) のバージョンは、4.0.13 以降である必要があります。VVR 4.0.13 は Apache Flink v1.13 をベースとしています。

• Lindorm インスタンスと Realtime Compute for Apache Flink のワークスペースが、同一の仮想プライベートクラウド (VPC) 内にあること。

  説明

  Realtime Compute for Apache Flink はデフォルトでインターネット経由でのアクセスが許可されていません。インターネット経由でデータを書き込む場合は、「Realtime Compute for Apache Flink によるインターネットへのアクセス方法」をご参照ください。

• Lindorm インスタンス上で LindormTSDB が有効化されていること。

• LindormTSDB バージョン 3.4.7 以降。バージョンの確認またはアップグレードについては、「LindormTSDB のリリースノート」および「Lindorm インスタンスのマイナーエンジンバージョンをアップグレードする」をご参照ください。

• Flink サービスの IP アドレスまたは CIDR ブロックが、Lindorm インスタンスのホワイトリストに追加されていること。Realtime Compute for Apache Flink の vSwitch の CIDR ブロックを取得するには、「ホワイトリストの設定方法」をご参照ください。Lindorm インスタンスのホワイトリストに IP アドレスまたは CIDR ブロックを追加するには、「ホワイトリストの設定」をご参照ください。

エンドツーエンドのフロー

Flink の結果を LindormTSDB に書き込む手順は以下のとおりです。

1. LindormTSDB シンクコネクタの JAR パッケージをダウンロードし、Realtime Compute for Apache Flink コンソールにアップロードします。

2. Flink SQL でソーステーブルおよび結果テーブルを定義します。結果テーブルの `WITH` 句を構成し、LindormTSDB を指すように設定します。

3. INSERT INTO 文を作成して、ソーステーブルのフィールドを結果テーブルにマップします。

4. Flink ジョブを送信します。このとき、コネクタが指定された LindormTSDB の時系列テーブルにデータポイントを書き込みます。

コネクタのセットアップ

LindormTSDB シンクコネクタ JAR パッケージをダウンロードし、Realtime Compute for Apache Flink コンソールにアップロードします。アップロード手順については、「JAR ドラフトの開発」をご参照ください。

結果テーブルの定義

Flink SQL で結果テーブルを作成し、`WITH` 句を使用して LindormTSDB の時系列テーブルにマップします。

CREATE TEMPORARY TABLE tsdb_sink(
  `timestamp` BIGINT,
  tag_<tagname> VARCHAR,
  field_<fieldname1> DOUBLE,
  field_<fieldname2> VARCHAR,
  field_<fieldname3> BIGINT,
  field_<fieldname4> BOOLEAN
  -- table VARCHAR (任意)
)
WITH (
    'connector' = 'lindormtsdb',
    'url' = '<lindormTSDBHttpUrl>',
    'table' = '<yourTableName>',
    'defaultDatabase' = '<yourDatabaseName>',
    'schemaPolicy' = '<schemaPolicy>',
    'sink.parallelism' = '<sinkParallelism>',
    'ignoreErrorData' = '<ignoreErrorData>',
    'maxRetries' = '<maxRetries>',
    'batchSize' = '<batchSize>',
    'connectTimeoutMs' = '<connectTimeoutMs>',
    'sync' = '<sync>',
    'debug' = '<debug>'
);

サンプル

以下のサンプルでは、DataGen ソースコネクタからデータを読み取り、mytable という名前の LindormTSDB 時系列テーブルに書き込みます。ソーステーブルは、id、score、name フィールドに対してランダム値を生成します。

-- ソーステーブル：DataGen コネクタを使用してランダムデータを生成
CREATE TEMPORARY TABLE datagen_source (
  id INTEGER,
  score DOUBLE,
  name STRING
)
WITH (
  'connector' = 'datagen'
);

-- 結果テーブル：LindormTSDB にマップ
-- schemaPolicy='weak' を指定すると、対象テーブルが存在しない場合に LindormTSDB が自動的に作成します。
-- 高スループットのジョブでは、sink.parallelism および batchSize の値を増加させることを推奨します。
CREATE TEMPORARY TABLE tsdb_sink(
  tag_tagk VARCHAR,
  field_score DOUBLE,
  field_name STRING,
  `timestamp` BIGINT
)
WITH (
    'connector' = 'lindormtsdb',
    'url' = 'http://ld-bp159jt4eivt3****-proxy-tsdb.lindorm.rds.aliyuncs.com:8242',
    'table' = 'mytable',
    'schemaPolicy' = 'weak'
    -- 'sink.parallelism' = '4',   -- 高スループットのワークロード向けに増加
    -- 'batchSize' = '1000'        -- 書き込みパフォーマンスを調整するためにバッチサイズを調整
);

-- ソーステーブルから LindormTSDB へデータを書き込み
INSERT INTO tsdb_sink
SELECT
  CAST(id AS STRING) AS tag_tagk,
  score AS field_score,
  name AS field_name,
  UNIX_TIMESTAMP(now()) * 1000 AS `timestamp`
FROM datagen_source;

結果の確認

Flink ジョブを送信した後、LindormTSDB へのデータ書き込みが正しく行われているかを確認します。

1. Realtime Compute for Apache Flink コンソールで、ジョブのステータスが 実行中 であることを確認します。

2. LindormTSDB に SQL コンソールまたはクライアントツールで接続し、以下のクエリを実行します。

   SELECT * FROM mytable LIMIT 10;

   クエリ結果として行が返された場合、書き込みパイプラインは正常に動作しています。

3. 行が返されない場合は、Flink ジョブのログを確認し、接続エラー、スキーマ検証失敗、またはタイムアウトメッセージがないかを確認します。

次のステップ

• スキーマの制約ポリシー

• LindormTSDB のリリースノート

• エンドポイントの表示