メインコンテンツまでスキップ
に最終更新

SeaTunnel

SeaTunnel は、使いやすく、非常に高い性能を持つ分散データ統合プラットフォームです。大量のデータのリアルタイム同期に対応し、毎日数百億件のレコードを安定的かつ効率的に同期できます。

このドキュメントは、SeaTunnel から Doris にデータを同期するユーザーを対象としています。ユースケースに応じた Doris Sink の選び方、パラメータの設定方法、使用方法を説明します。

コネクタの選び方​

ユースケース推奨する方法対象バージョンまたはエンジン説明
Exactly-once の書き込み、または CDC データ同期が必要な場合Connector-V2 Doris SinkSeaTunnel 2.3.1 以降Doris Sink、Exactly-once の書き込み、CDC データ同期に対応します。
Flink エンジン経由で Doris にデータを同期する場合Connector-V1 Flink Doris SinkSeaTunnel 2.1.0、Flink エンジンすでに Flink エンジンで実行している SeaTunnel ジョブに適しています。
Spark エンジン経由で Doris にデータを同期する場合Connector-V1 Spark Sink DorisSeaTunnel 2.1.0、Spark エンジンすでに Spark エンジンで実行しているジョブに適しています。たとえば、Hive から Doris へのデータ移行に使用できます。

シナリオ 1: Connector-V2 を使用して Doris に書き込む​

SeaTunnel 2.3.1 の Apache SeaTunnel Connector-V2 は、Doris Sink、Exactly-once の書き込み、CDC データ同期に対応しています。

プラグインのソースコードについては、SeaTunnel Doris Sink プラグインのコードを参照してください。

パラメータ設定​

パラメータ型必須デフォルト値説明
fenodesstringはい-Doris クラスターの FE ノードのアドレス。形式は fe_ip:fe_http_port,... です。
usernamestringはい-Doris のユーザー名。
passwordstringはい-Doris ユーザーのパスワード。
table.identifierstringはい-Doris のテーブル名。形式は DBName.TableName です。
sink.label-prefixstringはい-Stream Load で使用するラベルのプレフィックス。2PC を使用する場合、SeaTunnel の EOS セマンティクスを保証するために、このプレフィックスはグローバルに一意である必要があります。
sink.enable-2pcboolいいえtrue二段階コミット（2PC）を有効にするかどうか。デフォルトは true で、Exactly-once セマンティクスを保証します。二段階コミットについては、Stream Load マニュアルを参照してください。
sink.enable-deleteboolいいえfalse削除を有効にするかどうか。このオプションを使用するには、Doris テーブルでバッチ削除が有効になっている必要があります。Doris 0.15+ ではこの機能がデフォルトで有効で、Unique テーブルモデルのみがサポートされます。詳細については、バッチ削除を参照してください。
doris.configmapはい-Stream Load の data_desc パラメータ。その他のパラメータについては、Stream Load マニュアルを参照してください。

JSON 形式でデータをインポートする​

sink {
    Doris {
        fenodes = "doris_fe:8030"
        username = root
        password = ""
        table.identifier = "test.table_sink"
        sink.enable-2pc = "true"
        sink.label-prefix = "test_json"
        doris.config = {
            format = "json"
            read_json_by_line = "true"
        }
    }
}

CSV 形式でデータをインポートする​

sink {
    Doris {
        fenodes = "doris_fe:8030"
        username = root
        password = ""
        table.identifier = "test.table_sink"
        sink.enable-2pc = "true"
        sink.label-prefix = "test_csv"
        doris.config = {
            format = "csv"
            column_separator = ","
            line_delimiter = "\n"
        }
    }
}

シナリオ 2: Connector-V1 を使用して Doris に書き込む​

Apache SeaTunnel 2.1.0 は Doris コネクタに対応しており、Flink エンジンまたは Spark エンジンを使用して Doris にデータを同期できます。

プラグインのソースコードについては、SeaTunnel Flink Sink Doris プラグインのコードを参照してください。

パラメータ設定​

パラメータ型必須デフォルト値対応エンジン説明
fenodesstringはい-FlinkDoris FE の HTTP アドレス。例: 127.0.0.1:8030。
databasestringはい-FlinkDoris で書き込み先となるデータベースの名前。
tablestringはい-FlinkDoris で書き込み先となるテーブルの名前。
userstringはい-FlinkDoris へのアクセスに使用するユーザー。
passwordstringはい-FlinkDoris へのアクセスに使用するユーザーのパスワード。
batch_sizeintいいえ100Flink単一バッチで Doris に書き込む行数の上限。
intervalintいいえ1000Flinkフラッシュ間隔（ミリ秒）。この間隔が経過すると、非同期スレッドがバッファ内のデータを Doris に書き込みます。定期的な書き込みを無効にするには 0 を設定します。
max_retriesintいいえ1FlinkDoris への書き込みが失敗した後の再試行回数。
doris.*stringいいえ-FlinkStream Load のインポートパラメータ。例: doris.column_separator = ','。その他のパラメータについては、Stream Load マニュアルを参照してください。

例: Socket のデータを Doris に書き込む​

env {
    execution.parallelism = 1
}

source {
    SocketStream {
        host = 127.0.0.1
        port = 9999
        result_table_name = "socket"
        field_name = "info"
    }
}

transform {
}

sink {
    DorisSink {
        fenodes = "127.0.0.1:8030"
        user = root
        password = 123456
        database = test
        table = test_tbl
        batch_size = 5
        max_retries = 1
        interval = 5000
    }
}

ジョブを開始します。

sh bin/start-seatunnel-flink.sh --config config/flink.streaming.conf

Spark エンジンを使用して Doris に書き込む​

プラグインのソースコードについては、SeaTunnel Spark Sink Doris プラグインのコードを参照してください。

パラメータ設定​

パラメータ型必須デフォルト値対応エンジン説明
fenodesstringはい-SparkDoris FE ノードのアドレス。例: 127.0.0.1:8030。
databasestringはい-SparkDoris で書き込み先となるデータベースの名前。
tablestringはい-SparkDoris で書き込み先となるテーブルの名前。
userstringはい-SparkDoris へのアクセスに使用するユーザー。
passwordstringはい-SparkDoris へのアクセスに使用するユーザーのパスワード。
batch_sizeintはい100SparkSpark が Stream Load 経由で Doris に書き込むときに、バッチごとに送信する行数。
doris.*stringいいえ-SparkStream Load の HTTP パラメータ。Stream Load のパラメータを使用するには、doris. プレフィックスを付けます。例: doris.column_separator。その他のパラメータについては、Stream Load マニュアルを参照してください。

例: Hive から Doris にデータを移行する​

env {
    spark.app.name = "hive2doris-template"
}

spark {
    spark.sql.catalogImplementation = "hive"
}

source {
    hive {
        preSql = "select * from tmp.test"
        result_table_name = "test"
    }
}

transform {
}

sink {
    Console {
    }

    Doris {
        fenodes = "xxxx:8030"
        database = "tmp"
        table = "test"
        user = "root"
        password = "root"
        batch_size = 1000
        doris.column_separator = "\t"
        doris.columns = "date_key,date_value,day_in_year,day_in_month"
    }
}

ジョブを開始します。

sh bin/start-waterdrop-spark.sh --master local[4] --deploy-mode client --config ./config/spark.conf
Report issue