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

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

要点 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: PRECOMMITTED で停止し、その後の HTTP 呼び出しでコミットまたは中止する 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 タスクを送ります。対象となるすべての tablet の全レプリカから応答が届くと、トランザクションは VISIBLE になります。
  5. 読み取り。 VISIBLE になった後に実行されるクエリには新しい行が見えます。それより前に実行されたクエリには以前のバージョンが見えます。同じトランザクションに属する tablet 間で、一部だけが反映された状態を読み取ることはありません。

途中で何かが失敗した場合や、クライアントが 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 による重複排除を任せたいワークロード。

適さない用途

  • 高頻度の単一行挿入。各トランザクションでコミットの往復通信と tablet のバージョンが増え、小さなデータロードはコミット段階がボトルネックになります。代わりに Group Commit を使ってください。
  • 複数のデータベースにまたがる原子性。一つのトランザクション内のすべてのテーブルは、同じデータベースに属している必要があります。
  • クラウド／ストレージとコンピュートを分離した構成の Merge-on-Write Unique Key テーブルに対する Stream Load 2PC。BE はこの組み合わせを Status::NotSupported で拒否します。オンプレミスの shared-nothing クラスターでは利用できます。
  • 長期間にわたる重複排除。Label は約 3 日が経過するか、データベースあたり 2,000 件に達すると削除されます。先に達した条件が適用され、その後に同じ Label を再利用しても重複は排除されません。長期間保持するべき等性キーは、アプリケーション側で管理してください。
  • 同じトランザクション内で自分の書き込みを読み取ること。各文は開始時点のスナップショットに対して実行されるため、前の INSERT の結果は COMMIT の後まで見えません。

関連資料​

Report issue