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

データロードトランザクション

要約 Apache Doris は、すべてのデータロードをトランザクションで処理します。複数の挿入を一つの BEGIN ... COMMIT にまとめ、データロードの Label で再試行時の重複を排除できます。また、Stream Load の HTTP リクエストに指定するヘッダーを使って二段階コミットを実行できます。この仕組みにより、Flink、Spark、CDC パイプライン は、独自の管理用テーブルを持たずに Doris へ exactly-once で書き込めます。

Apache Doris のデータロードトランザクション：複数のステートメントによるデータロードを単一のアトミックな単位として実行し、Label による重複排除と 2PC でストリーミングシンクの exactly-once を実現します。

Apache Doris でデータロードトランザクションを使う理由​

Apache Doris のデータロードトランザクションは、行の重複や書き込みの一部だけが完了する問題をデータウェアハウス側で処理します。これにより、Flink の再実行、途中で止まった複数テーブルへの書き込み、ファイルの再アップロードにも適切に対処できます。データウェアハウスへのデータロードでは、よくある問題が起こります。Flink ジョブがチェックポイントの途中で再起動して直前のバッチを再実行し、行が二重に登録されます。CDC パイプラインがファクトテーブルを更新できても、対応するディメンションテーブルへの書き込みに失敗し、次のバックフィルまで両者に不整合が残ることがあります。また、前回の試行が実際には成功したと分からないため、再試行スクリプトが同じファイルを再アップロードすることもあります。いずれもアプリケーション側に煩雑な管理処理を強います。

  • 停滞したストリーミングジョブの再実行で、行が重複します。
  • 一緒に反映すべき複数テーブルへの書き込みが、途中で止まることがあります。
  • 同じファイルを再送信すると、アプリケーション側で追跡していない限り、二重に取り込まれます。

Apache Doris はこれら三つをサーバー側で処理するため、データロードクライアントをシンプルに保てます。

Apache Doris のデータロードトランザクションとは​

Apache Doris のデータロードトランザクションは、すべてのデータ取り込み経路を支えるアトミックな書き込み単位です。Doris への書き込みはすべてトランザクション内で実行されます。対象には INSERT INTO、Stream Load、Routine Load、Broker Load が含まれます。各トランザクションには一意の Label が割り当てられ、FE が管理する一連のライフサイクル状態を経て進みます。Apache Doris にデータロードごとのトランザクションの開始とコミットを暗黙的に任せることも、明示的な BEGIN/COMMIT/ROLLBACK で制御することもできます。ストリーミングシンクでは、HTTP 経由でコミットを二段階に分けることもできます。

主な用語

  • Label：データベース内でトランザクションを識別する文字列です。同じ Label で再送信すると、データロードの重複が排除されます。
  • Implicit transaction：個々のデータロードがそれぞれ独立したトランザクションとなり、成功時に自動でコミットされます。
  • Explicit transaction：BEGIN ... COMMIT; により、同じデータベースへの複数の書き込みを一つのアトミックな単位にまとめます。
  • Stream Load 2PC：後続の HTTP 呼び出しでコミットまたは中止するまで、PRECOMMITTED で停止する Stream Load です。
  • Lifecycle states：PREPARE → PRECOMMITTED → COMMITTED → VISIBLE（または ABORTED）です。行をクエリで参照できるのは VISIBLE になってからです。

Apache Doris のデータロードトランザクションの仕組み​

Apache Doris のデータロードトランザクションでは、すべての書き込みが FE の管理する五つの状態（PREPARE → PRECOMMITTED → COMMITTED → VISIBLE、失敗時は ABORTED）を経ます。Label によって再試行時の重複を排除し、最後にレプリカ単位で公開することで、ある一時点から行をクエリで参照できるようにします。

  1. 開始。 データロードは PREPARE 状態でトランザクションを開始します。FE は TxnId を割り当て、ユーザーが指定した Label または生成された Label に関連付けます。同じ Label で再度送信すると、元のトランザクションの結果が返されます。
  2. 書き込み。 BE がデータを受け取り、セグメントを構築して成功を報告します。予定されたすべてのデータロードから完了通知が届くと、トランザクションは PRECOMMITTED に進みます。
  3. コミット。 COMMIT（または第二段階の txn_operation: commit HTTP 呼び出し）によって、トランザクションは COMMITTED に進みます。データはディスク上にありますが、まだ読み取れません。
  4. 公開。 バックグラウンドデーモンが各レプリカに PublishVersion タスクを送ります。対象となるすべてのタブレットの全レプリカから完了通知が届くと、トランザクションは VISIBLE になります。
  5. 読み取り。 VISIBLE になった後に実行されるクエリには新しい行が見えます。それ以前のクエリには前のバージョンが見えます。同じトランザクション内のタブレット間で、更新の一部だけが見える読み取りは発生しません。

途中で何かが失敗した場合や、クライアントが ROLLBACK（または txn_operation: abort）を呼び出した場合、トランザクションは ABORTED となり、データは破棄されます。

クイックスタート​

CREATE TABLE orders (
  id INT, amount DECIMAL(10,2)
) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 4;

CREATE TABLE order_audit (
  id INT, action STRING
) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 4;

BEGIN;
INSERT INTO orders       VALUES (1, 99.50), (2, 149.00);
INSERT INTO order_audit  VALUES (1, 'created'), (2, 'created');
COMMIT;

期待される結果

Query OK, 2 rows affected
{'label':'txn_insert_b55db21aad7451b','status':'VISIBLE','txnId':'10013'}

両方のテーブルが同時に読み取り可能になります。二つ目の INSERT が失敗した場合や、COMMIT 前に接続が切れた場合は、どちらのテーブルにも新しい行は反映されません。

Apache Doris のデータロードトランザクションを使う場面​

複数の挿入をまとめて反映する必要がある場合、ストリーミングシンクに exactly-once セマンティクスが必要な場合、または再試行されたデータロードをアプリケーション側ではなく Label で重複排除したい場合に、Apache Doris のデータロードトランザクションを使います。

適した用途

  • チェックポイントをまたいだ exactly-once セマンティクスのために Stream Load 2PC を使用する Flink や Spark のシンク。
  • ファクトテーブルとディメンションテーブルをまとめて更新する必要がある CDC パイプライン。
  • 関連する行のバッチを一つの単位として追加または削除するバックフィル。
  • 冪等な再試行が重要で、クライアントではなくデータウェアハウス側で Label による重複排除を行いたいワークロード。

適さない用途

  • 高頻度の単一行挿入。トランザクションごとにコミットの往復通信とタブレットのバージョンが増え、小規模なデータロードではコミット段階がボトルネックになります。代わりに Group Commit を使用してください。
  • 二つのデータベースにまたがるアトミックな処理。一つのトランザクションに含まれるすべてのテーブルは、同じデータベースに属している必要があります。
  • cloud / storage-compute-separated 構成での Merge-on-Write Unique Key テーブルに対する Stream Load 2PC。この組み合わせは BE によって Status::NotSupported で拒否されます。on-prem (shared-nothing) クラスターでは使用できます。
  • 長期間にわたる重複排除。Label は約3日後、またはデータベースあたり2,000件に達した時点のうち、早い方で削除されます。その後に同じ Label を再利用しても重複は排除されません。長期間有効な冪等性キーはアプリケーション側で追跡してください。
  • 同じトランザクション内で自分の書き込みを読み取る処理。各ステートメントは開始時点のスナップショットに対して実行されるため、先行する INSERT の結果は COMMIT 後まで見えません。

関連資料​

Report issue