DataX を使用してリレーショナルデータベースから LindormTSDB にデータを移行する - Lindorm

このトピックでは、DataX を使用してリレーショナルデータベースから LindormTSDB にデータを移行する方法について説明します。DataX は、Alibaba Group によって提供されるオープンソースツールです。

背景情報

DataX は、Alibaba Group 内で広く使用されているオフラインデータ同期ツールです。 DataX を使用すると、MySQL、Oracle、SQL Server、PostgreSQL、Hadoop Distributed File System（HDFS）、Hive、AnalyticDB for MySQL、HBase、Tablestore（OTS）、MaxCompute、PolarDB-X など、さまざまな異なるデータソース間でデータを効率的に同期できます。MaxCompute は、以前は Open Data Processing Service（ODPS）として知られていました。 DataX の使用方法の詳細については、README をご参照ください。

以下のセクションでは、DataX について説明し、リレーショナルデータベースから LindormTSDB のデータベースにデータを移行する方法の例を示します。この例では、MySQL データベースが使用されています。データ移行を実行するには、MySQL Reader と TSDB Writer を使用します。

MySQL Reader

MySQL Reader は、DataX によって提供されるプラグインです。 MySQL Reader を使用すると、MySQL データベースからデータをクエリできます。

TSDB Writer

DataX は、LindormTSDB のデータベースにデータを書き込むために使用される TSDB Writer プラグインを提供します。

注意事項

DataX が LindormTSDB にアクセスできることを確認してください。

TSDB Writer は、/api/mput HTTP エンドポイントと /api/put HTTP エンドポイントを呼び出してデータを書き込みます。データを移行する場合は、移行タスクの各プロセスが LindormTSDB によって提供される HTTP エンドポイントにアクセスできることを確認してください。アクセスできない場合、接続例外がスローされます。

移行タスクの各プロセスが MySQL データベースにアクセスできることを確認してください。

MySQL Reader は、Java Database Connectivity（JDBC）を使用してデータを読み取ります。データを移行する必要がある場合は、移行タスクの各プロセスが MySQL データベースによって提供される JDBC API にアクセスできることを確認してください。アクセスできない場合、接続例外がスローされます。

準備

環境を構成します。
- Linux
- Java Development Kit（JDK）: JDK 1.8 以降を使用します。
- Python: Python 2.6.x を使用することをお勧めします。
- MySQL: DataX は MySQL 5.x のみに対応しています。
- LindormTSDB: LindormTSDB をアクティブ化します。
DataX とプラグインをダウンロードします。
ここをクリックして、対応する TAR ファイルをダウンロードします。
DataX によって提供される組み込みスクリプトを使用して、データが想定どおりに移行できるかどうかをテストします。
テストで使用されるプラグインは、Stream Reader と Stream Writer です。 Stream Reader と Stream Writer は外部依存関係を必要としません。そのため、Stream Reader と Stream Writer はテストに適しています。 Stream Reader と Stream Writer は、単純なデータ移行プロセスをシミュレートするために使用されます。 Stream Reader はランダムな文字列を生成します。 Stream Writer は文字列を受信し、CLI に表示します。

DataX のインストール

インストールパッケージを DATAX_HOME に解凍します。次に、次のコマンドを実行して、移行テストタスクを開始します。

$ cd ${DATAX_HOME}
$ python bin/datax.py job/job.json
// ${DATAX_HOME} に移動します。
// job/job.json ファイルを使用して DataX を実行します。

移行結果の確認

次の例は、データが想定どおりに移行された場合に返される概要情報を示しています。

Task start time                    : 2019-04-26 11:18:07
Task end time                    : 2019-04-26 11:18:17
Execution time                    :                 10s
Average traffic                    :          253.91KB/s
Write rate                    :          10000rec/s
Records obtained                    :              100000
Number of write and read failures                    :                   0
// タスクの開始時刻
// タスクの終了時刻
// 実行時間
// 平均トラフィック
// 書き込みレート
// 取得したレコード数
// 書き込みと読み取りの失敗数

MySQL データベースから LindormTSDB のデータベースにデータを移行するタスクの構成と実行

前の手順では、Stream Reader と Stream Writer を使用すると、データが想定どおりに移行されます。移行結果から、DataX を使用してデータを移行できることがわかります。この場合、MySQL Reader と TSDB Writer を使用して、MySQL データベースから LindormTSDB のデータベースにデータを移行できます。

パラメーター

次の表に、構成できるパラメーターを示します。

MySQL Reader パラメーター

パラメーター	タイプ	必須	説明	デフォルト値	例
jdbcUrl	String	はい	JDBC が MySQL データベースに接続するために使用するデータベース URL を指定します。	なし	`jdbc:mysql://127.0.0.1:3306/datax`
username	String	はい	MySQL データベースに接続するために使用するユーザー名を指定します。	なし	`root`
password	String	はい	MySQL データベースに接続するために使用するパスワードを指定します。	なし	`root`
table	String	はい	移行するデータを含むテーブルを指定します。	なし	`book`
column	Array	はい	移行する列の名前を指定します。	`[]`	`["m"]`
splitPk	String	いいえ	テーブルシャーディングに使用されるシャードキーを指定します。 splitPk に値を指定すると、指定された値に基づいてテーブルがシャーディングされます。	なし	`id`

TSDB Writer パラメーター

パラメーター	タイプ	必須	説明	デフォルト値	例
sourceDbType	String	いいえ	有効な値: TSDB と RDB。 TSDB の値は、OpenTSDB、InfluxDB、Prometheus、TimeScale などの時系列エンジンを指定します。 RDB の値は、MySQL、Oracle、PostgreSQL、PolarDB-X などのリレーショナルデータベースエンジンを指定します。	`TSDB`	`RDB`
column	Array	はい	移行する列の名前を指定します。	`[]`	`["name","type","create_time","price"]`
columnType	Array	はい	列が移行された後、LindormTSDB のデータベース内の列のデータ型を指定します。有効な値: timestamp、tag、metric_num、metric_string。[タイムスタンプ] 値は、列にタイムスタンプが含まれていることを指定します。[タグ] 値は、列にタグが含まれていることを指定します。[metric_num] 値は、列に数値データ型のデータが含まれていることを指定します。[metric_string] 値は、列に文字列データ型のデータが含まれていることを指定します。	`[]`	`["tag","tag","timestamp","metric_num"]`
endpoint	String	はい	LindormTSDB のエンドポイントを指定します。値は IP:PORT 形式で指定します。	なし	`ld-xxxx:8242`
multiField	Bool	はい	/api/mput HTTP エンドポイントを呼び出してデータ書き込み操作を実行するかどうかを指定します。値を true に設定することをお勧めします。	false	true
table	String	はい	LindormTSDB のデータベースに移行するテーブルを指定します。	なし	table
username	String	いいえ	LindormTSDB のデータベースに接続するために使用するユーザー名を指定します。	なし	tsdbuser
password	String	いいえ	LindormTSDB のデータベースに接続するために使用するパスワードを指定します。	なし	tsdbpsw
ignoreWriteError	Bool	はい	書き込みエラーを無視するかどうかを指定します。	false	true
database	String	いいえ	データを書き込むデータベースの名前を指定します。	default	default

データ移行タスクの構成

MySQL データベースから LindormTSDB のデータベースにデータを移行するタスクを構成します。この例では、タスク名は mysql2tsdb.json です。次のコードは、データ移行タスクを構成する方法の例を示しています。パラメーターの詳細については、パラメーターをご参照ください。

{
  "job": {
    "content": [
      {
        "reader": {
          "name": "mysqlreader",
          "parameter": {
            "column": [
              // 移行するカラム名
              "name",
              "type",
              "create_time",
              "price",
              "description",
              "available"
            ],
            "connection": [
              {
                "jdbcUrl": [
                  // MySQL データベースの JDBC URL
                  "jdbc:mysql://127.0.0.1:3306/datax"
                ],
                "table": [
                  // 移行元テーブル名
                  "book"
                ]
              }
            ],
            "password": "xxx", // MySQL データベースのパスワード
            "splitPk": "id", // テーブルシャーディングに使用するキー
            "username": "root" // MySQL データベースのユーザー名
          }
        },
        "writer": {
          "name": "tsdbwriter",
          "parameter": {
            "column": [
              // 移行するカラム名
              "name",
              "type",
              "create_time",
              "price",
              "description",
              "available"
            ],
            "columnType": [
              // LindormTSDB でのカラムのデータ型
              "tag",
              "tag",
              "timestamp",
              "field_double",
              "field_string",
              "field_bool"
            ],
            "sourceDbType": "RDB", // ソースデータベースのタイプ
            "endpoint": "http://localhost:8242", // LindormTSDB のエンドポイント
            "table": "my_book", // 移行先テーブル名
            "multiField": "true", // /api/mput を使用して書き込むかどうか
            "username":"xxx", // LindormTSDB のユーザー名
            "password":"xxx", // LindormTSDB のパスワード
            "ignoreWriteError":"false", // 書き込みエラーを無視するかどうか
            "database":"default" // LindormTSDB のデータベース名
          }
        }
      }
    ],
    "setting": {
      "speed": {
        "channel": 1 // 並列チャネル数
      }
    }
  }
}

一貫した列名のシーケンス

TSDB Writer で構成された列名のシーケンスが、MySQL Reader で構成された列名のシーケンスと同じであることを確認してください。同じでない場合、データは想定どおりに処理されません。

タスクの開始

$ cd ${DATAX_HOME}/..
$ ls
  datax/  datax.tar.gz  mysql2tsdb.json
$ python datax/bin/datax.py mysql2tsdb.json
// datax ディレクトリの親ディレクトリに移動します。
// ファイル一覧を表示します。
// mysql2tsdb.json ファイルを使用して DataX を実行します。

移行結果の確認

次の例は、データが想定どおりに移行された場合に返される概要情報を示しています。

Task start time                    : 2019-05-21 18:25:16
Task end time                    : 2019-05-21 18:25:27
Execution time                    :                 11s
Average traffic                    :                3B/s
Write rate                    :               0rec/s
Records obtained                    :                   3
Number of write and read failures                    :                   0
// タスクの開始時刻
// タスクの終了時刻
// 実行時間
// 平均トラフィック
// 書き込みレート
// 取得したレコード数
// 書き込みと読み取りの失敗数

FAQ

[移行プロセスで Java Virtual Machine（jvm）のメモリサイズを変更できますか？]

はい、移行プロセスの JVM メモリサイズを変更できます。 MySQL データベースから LindormTSDB のデータベースにデータを移行するタスクの JVM メモリサイズを調整する場合は、次のコマンドを実行できます。

python datax/bin/datax.py mysql2tsdb.json -j "-Xms4096m -Xmx4096m"    // JVM の初期メモリサイズと最大メモリサイズを指定して DataX を実行します。

[lindormtsdb の IP アドレスホワイトリストを構成するにはどうすればよいですか？]

詳細については、「ホワイトリストの構成」をご参照ください。

[移行タスクが Elastic Compute Service（ecs）インスタンスで実行されている場合、仮想プライベートクラウド（vpc）を構成するにはどうすればよいですか？また、一般的な問題は何ですか？]

詳細については、「異なるシナリオでのセキュリティグループの構成」および「よくある質問」をご参照ください。