メインコンテンツまでスキップ

DataX Doriswriter

DataX Doriswriterプラグインは、MySQL、Oracle、SQL サーバーなどの様々なデータソースからStream Load方式を使用してDorisにデータを同期することをサポートします。

Note

このプラグインはDataXサービスと併せて使用する必要があります。 DataXは複数のデータソースをサポートしています。詳細については、こちらを参照してください。

使用方法​

DataXインストールパッケージを直接ダウンロード​

DataXは、DataXが既に含まれた公式インストールパッケージを提供しており、直接ダウンロードして使用することができます。詳細については、こちらを参照してください。

DorisWriterプラグインを手動でコンパイル​

DorisWriterプラグインのソースコードをダウンロードします。

  1. init-env.shを実行

  2. 必要に応じてDataX/doriswriterのdoriswriterのコードを修正

  3. doriswriterをビルド

    doriswriterを単独でビルド:

     `mvn clean install -pl plugin-rdbms-util,doriswriter -DskipTests`

    DataXプロジェクト全体をコンパイルする必要がある場合は、こちらを参照してください

    コンパイルエラー

     以下のコンパイルエラーが発生した場合:

     ```
     Could not find artifact com.alibaba.datax:datax-all:pom:0.0.1-SNAPSHOT ...
     ```

以下のソリューションを試すことができます：

    1. [alibaba-datax-maven-m2-20210928.tar.gz](https://doris-thirdparty-repo.bj.bcebos.com/thirdparty/alibaba-datax-maven-m2-20210928.tar.gz) をダウンロードする
    2. 解凍後、生成された `alibaba/datax/` ディレクトリを、使用するmavenに対応する `.m2/repository/com/alibaba/` にコピーし、再度コンパイルを試行する。

Datax DorisWriter パラメータ説明：​

  • jdbcUrl

    • 説明: DorisのJDBC接続文字列、ユーザーはpreSqlまたはpostSQLを実行します。
    • 必須: はい
    • デフォルト: なし

  • loadUrl

    • 説明: Stream Loadの接続先として使用します。形式は "ip:port" です。IPはFEノードのIP、portはFEノードのhttp_portです。複数指定可能で、英語のカンマ , で区切ります。doriswriterはポーリング方式でアクセスします。
    • 必須: はい
    • デフォルト: なし

  • username

    • 説明: Dorisデータベースにアクセスするためのユーザー名
    • 必須: はい
    • デフォルト: なし

  • password

    • 説明: Dorisデータベースにアクセスするためのパスワード
    • 必須: いいえ
    • デフォルト: 空

  • connection.selectedDatabase

    • 説明: 書き込みが必要なDorisデータベースの名前。
    • 必須: はい
    • デフォルト: なし

  • connection. table

    • 説明: 書き込みが必要なDorisテーブルの名前。
      • 必須: はい
      • デフォルト: なし

  • flushInterval

    • 説明: データをバッチ書き込みする時間間隔。この時間間隔を小さく設定しすぎるとDorisの書き込みブロッキング問題が発生し、エラーコード-235になります。この時間間隔を小さく設定し、maxBatchRowsとbatchSizeパラメータを大きく設定した場合、設定したデータサイズに達しなくてもインポートされる可能性があります。
    • 必須: いいえ
    • デフォルト: 30000 (ms)

  • column

    • 説明: 宛先テーブルにデータを書き込む必要があるフィールド。これらのフィールドは生成されるJSONデータのフィールド名として使用されます。フィールドはカンマで区切ります。例: "column": ["id","name","age"]。
    • 必須: はい
    • デフォルト: なし

  • preSql

    • 説明: 宛先テーブルにデータを書き込む前に、ここの標準ステートメントが最初に実行されます。
    • 必須: いいえ
    • デフォルト: なし

  • postSql

    • 説明: 宛先テーブルにデータを書き込んだ後に、ここの標準ステートメントが実行されます。
    • 必須: いいえ
    • デフォルト: なし

  • maxBatchRows

    • 説明: 各バッチでインポートするデータの最大行数。batchSizeと合わせて、バッチあたりのインポートレコード行数を制御します。各バッチのデータが2つの閾値のいずれかに達すると、そのバッチのデータのインポートが開始されます。
    • 必須: いいえ
    • デフォルト: 500000

  • batchSize

    • 説明: 各バッチでインポートするデータの最大量。maxBatchRowsと連携して、バッチあたりのインポート数を制御します。各バッチのデータが2つの閾値のいずれかに達すると、そのバッチのデータのインポートが開始されます。
    • 必須: いいえ
    • デフォルト: 94371840

  • maxRetries

    • 説明: 各バッチのデータインポート失敗後のリトライ回数。
    • 必須: いいえ
    • デフォルト: 3

  • labelPrefix

    • 説明: 各バッチのインポートタスクのlabelプレフィックス。最終的なlabelは labelPrefix + UUID で構成され、グローバルに一意なlabelを形成してデータの重複インポートを防ぎます
    • 必須: いいえ
    • デフォルト: datax_doris_writer_

  • loadProps

    • 説明: StreamLoadのリクエストパラメータ。詳細については、StreamLoad紹介ページを参照してください。Stream load - Apache Doris

      これにはインポートするデータ形式: format等が含まれます。インポートするデータ形式はデフォルトでcsvで、JSONをサポートします。詳細については、下記の型変換セクションを参照するか、上記のStream loadの公式情報を参照してください。

    • 必須: いいえ

    • デフォルト: なし

例​

1. Streamがデータを読み込み、Dorisにインポートする​

doriswriterプラグインの使用方法については、こちらを参照してください。

2.Mysqlがデータを読み込み、Dorisにインポートする​

1.Mysqlテーブル構造

CREATE TABLE `t_test`(
 `id`bigint(30) NOT NULL,
 `order_code` varchar(30) DEFAULT NULL COMMENT '',
 `line_code` varchar(30) DEFAULT NULL COMMENT '',
 `remark` varchar(30) DEFAULT NULL COMMENT '',
 `unit_no` varchar(30) DEFAULT NULL COMMENT '',
 `unit_name` varchar(30) DEFAULT NULL COMMENT '',
 `price` decimal(12,2) DEFAULT NULL COMMENT '',
 PRIMARY KEY(`id`) USING BTREE
)ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 ROW_FORMAT=DYNAMIC COMMENT='';

2.Dorisテーブル構造

CREATE TABLE `ods_t_test` (
 `id`bigint(30) NOT NULL,
 `order_code` varchar(30) DEFAULT NULL COMMENT '',
 `line_code` varchar(30) DEFAULT NULL COMMENT '',
 `remark` varchar(30) DEFAULT NULL COMMENT '',
 `unit_no` varchar(30) DEFAULT NULL COMMENT '',
 `unit_name` varchar(30) DEFAULT NULL COMMENT '',
 `price` decimal(12,2) DEFAULT NULL COMMENT ''
）ENGINE=OLAP
UNIQUE KEY(id`, `order_code`)
DISTRIBUTED BY HASH(`order_code`) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 3",
"in_memory" = "false",
"storage_format" = "V2"
);

3.dataxスクリプトを作成する

my_import.json

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "column": ["id","order_code","line_code","remark","unit_no","unit_name","price"],
                        "connection": [
                            {
                                "jdbcUrl": ["jdbc:mysql://localhost:3306/demo"],
                                "table": ["employees_1"]
                            }
                        ],
                        "username": "root",
                        "password": "xxxxx",
                        "where": ""
                    }
                },
                "writer": {
                    "name": "doriswriter",
                    "parameter": {
                        "loadUrl": ["127.0.0.1:8030"],
                        "column": ["id","order_code","line_code","remark","unit_no","unit_name","price"],
                        "username": "root",
                        "password": "xxxxxx",
                        "postSql": ["select count(1) from all_employees_info"],
                        "preSql": [],
                        "flushInterval":30000,
                        "connection": [
                          {
                            "jdbcUrl": "jdbc:mysql://127.0.0.1:9030/demo",
                            "selectedDatabase": "demo",
                            "table": ["all_employees_info"]
                          }
                        ],
                        "loadProps": {
                            "format": "json",
                            "strip_outer_array":"true",
                            "line_delimiter": "\\x02"
                        }
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": "1"
            }
        }
    }
}

備考:

"loadProps": {
  "format": "json",
  "strip_outer_array": "true",
  "line_delimiter": "\\x02"
}
  1. ここではJSON形式を使用してデータをインポートします
  2. line_delimiterはデフォルトで改行文字ですが、データ内の値と競合する可能性があるため、特殊文字や不可視文字を使用してインポートエラーを回避できます
  3. strip_outer_array : インポートデータのバッチ内の複数行のデータを表します。Dorisは解析時に配列を展開し、その中の各Objectを順番に1行のデータとして解析します。
  4. Stream loadのその他のパラメータについては、Stream load - Apache Dorisを参照してください
  5. CSV形式の場合は、次のように使用できます
"loadProps": {
   "format": "csv",
   "column_separator": "\\x01",
   "line_delimiter": "\\x02"
}

CSV形式では、データ内の特殊文字との競合を避けるために行と列の区切り文字に特に注意する必要があります。ここでは隠し文字を推奨します。デフォルトの列区切り文字は: \t、行区切り文字: \n

4.dataxタスクを実行します。詳細はdatax公式サイトを参照してください

python bin/datax.py my_import.json

実行後、以下の情報を確認できます

2022-11-16 14:28:54.012 [job-0] INFO  JobContainer - jobContainer starts to do prepare ...
2022-11-16 14:28:54.012 [job-0] INFO  JobContainer - DataX Reader.Job [mysqlreader] do prepare work .
2022-11-16 14:28:54.013 [job-0] INFO  JobContainer - DataX Writer.Job [doriswriter] do prepare work .
2022-11-16 14:28:54.020 [job-0] INFO  JobContainer - jobContainer starts to do split ...
2022-11-16 14:28:54.020 [job-0] INFO  JobContainer - Job set Channel-Number to 1 channels.
2022-11-16 14:28:54.023 [job-0] INFO  JobContainer - DataX Reader.Job [mysqlreader] splits to [1] tasks.
2022-11-16 14:28:54.023 [job-0] INFO  JobContainer - DataX Writer.Job [doriswriter] splits to [1] tasks.
2022-11-16 14:28:54.033 [job-0] INFO  JobContainer - jobContainer starts to do schedule ...
2022-11-16 14:28:54.036 [job-0] INFO  JobContainer - Scheduler starts [1] taskGroups.
2022-11-16 14:28:54.037 [job-0] INFO  JobContainer - Running by standalone Mode.
2022-11-16 14:28:54.041 [taskGroup-0] INFO  TaskGroupContainer - taskGroupId=[0] start [1] channels for [1] tasks.
2022-11-16 14:28:54.043 [taskGroup-0] INFO  Channel - Channel set byte_speed_limit to -1, No bps activated.
2022-11-16 14:28:54.043 [taskGroup-0] INFO  Channel - Channel set record_speed_limit to -1, No tps activated.
2022-11-16 14:28:54.049 [taskGroup-0] INFO  TaskGroupContainer - taskGroup[0] taskId[0] attemptCount[1] is started
2022-11-16 14:28:54.052 [0-0-0-reader] INFO  CommonRdbmsReader$Task - Begin to read record by Sql: [select taskid,projectid,taskflowid,templateid,template_name,status_task from dwd_universal_tb_task 
] jdbcUrl:[jdbc:mysql://localhost:3306/demo?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
Wed Nov 16 14:28:54 GMT+08:00 2022 WARN: Establishing SSL connection without server's identity verification is not recommended. According to MySQL 5.5.45+, 5.6.26+ and 5.7.6+ requirements SSL connection must be established by default if explicit option isn't set. For compliance with existing applications not using SSL the verifyServerCertificate property is set to 'false'. You need either to explicitly disable SSL by setting useSSL=false, or set useSSL=true and provide truststore for server certificate verification.
2022-11-16 14:28:54.071 [0-0-0-reader] INFO  CommonRdbmsReader$Task - Finished read record by Sql: [select taskid,projectid,taskflowid,templateid,template_name,status_task from dwd_universal_tb_task 
] jdbcUrl:[jdbc:mysql://localhost:3306/demo?yearIsDateType=false&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false&rewriteBatchedStatements=true].
2022-11-16 14:28:54.104 [Thread-1] INFO  DorisStreamLoadObserver - Start to join batch data: rows[2] bytes[438] label[datax_doris_writer_c4e08cb9-c157-4689-932f-db34acc45b6f].
2022-11-16 14:28:54.104 [Thread-1] INFO  DorisStreamLoadObserver - Executing stream load to: 'http://127.0.0.1:8030/api/demo/dwd_universal_tb_task/_stream_load', size: '441'
2022-11-16 14:28:54.224 [Thread-1] INFO  DorisStreamLoadObserver - StreamLoad response :{"Status":"Success","BeginTxnTimeMs":0,"Message":"OK","NumberUnselectedRows":0,"CommitAndPublishTimeMs":17,"Label":"datax_doris_writer_c4e08cb9-c157-4689-932f-db34acc45b6f","LoadBytes":441,"StreamLoadPutTimeMs":1,"NumberTotalRows":2,"WriteDataTimeMs":11,"TxnId":217056,"LoadTimeMs":31,"TwoPhaseCommit":"false","ReadDataTimeMs":0,"NumberLoadedRows":2,"NumberFilteredRows":0}
2022-11-16 14:28:54.225 [Thread-1] INFO  DorisWriterManager - Async stream load finished: label[datax_doris_writer_c4e08cb9-c157-4689-932f-db34acc45b6f].
2022-11-16 14:28:54.249 [taskGroup-0] INFO  TaskGroupContainer - taskGroup[0] taskId[0] is successed, used[201]ms
2022-11-16 14:28:54.250 [taskGroup-0] INFO  TaskGroupContainer - taskGroup[0] completed it's tasks.
2022-11-16 14:29:04.048 [job-0] INFO  StandAloneJobContainerCommunicator - Total 2 records, 214 bytes | Speed 21B/s, 0 records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 0.000s |  All Task WaitReaderTime 0.000s | Percentage 100.00%
2022-11-16 14:29:04.049 [job-0] INFO  AbstractScheduler - Scheduler accomplished all tasks.
2022-11-16 14:29:04.049 [job-0] INFO  JobContainer - DataX Writer.Job [doriswriter] do post work.
Wed Nov 16 14:29:04 GMT+08:00 2022 WARN: Establishing SSL connection without server's identity verification is not recommended. According to MySQL 5.5.45+, 5.6.26+ and 5.7.6+ requirements SSL connection must be established by default if explicit option isn't set. For compliance with existing applications not using SSL the verifyServerCertificate property is set to 'false'. You need either to explicitly disable SSL by setting useSSL=false, or set useSSL=true and provide truststore for server certificate verification.
2022-11-16 14:29:04.187 [job-0] INFO  DorisWriter$Job - Start to execute preSqls:[select count(1) from dwd_universal_tb_task]. context info:jdbc:mysql://172.16.0.13:9030/demo.
2022-11-16 14:29:04.204 [job-0] INFO  JobContainer - DataX Reader.Job [mysqlreader] do post work.
2022-11-16 14:29:04.204 [job-0] INFO  JobContainer - DataX jobId [0] completed successfully.
2022-11-16 14:29:04.204 [job-0] INFO  HookInvoker - No hook invoked, because base dir not exists or is a file: /data/datax/hook
2022-11-16 14:29:04.205 [job-0] INFO  JobContainer - 
         [total cpu info] => 
                averageCpu                     | maxDeltaCpu                    | minDeltaCpu                    
                -1.00%                         | -1.00%                         | -1.00%
                        

         [total gc info] => 
                 NAME                 | totalGCCount       | maxDeltaGCCount    | minDeltaGCCount    | totalGCTime        | maxDeltaGCTime     | minDeltaGCTime     
                 PS MarkSweep         | 1                  | 1                  | 1                  | 0.017s             | 0.017s             | 0.017s             
                 PS Scavenge          | 1                  | 1                  | 1                  | 0.007s             | 0.007s             | 0.007s             

2022-11-16 14:29:04.205 [job-0] INFO  JobContainer - PerfTrace not enable!
2022-11-16 14:29:04.206 [job-0] INFO  StandAloneJobContainerCommunicator - Total 2 records, 214 bytes | Speed 21B/s, 0 records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 0.000s |  All Task WaitReaderTime 0.000s | Percentage 100.00%
2022-11-16 14:29:04.206 [job-0] INFO  JobContainer - 
Task Start Time                        : 2022-11-16 14:28:53
Task End Time                          : 2022-11-16 14:29:04
Total Task Duration                    : 10s
Average Task Throughput                : 21B/s
Record Write Speed                     : 0rec/s
Total Records Read                     : 2
Total Read/Write Failures              : 0

Report issue
ASF
Resources
Community
Join the community
Copyright © 2026 The Apache Software Foundation,Licensed under the Apache License, Version 2.0. Apache, Doris, Apache Doris, the Apache feather logo and the Apache Doris logo are trademarks of The Apache Software Foundation.