ロードトランザクション
要点 Apache Doris では、すべてのデータロードがトランザクション内で実行されます。複数の挿入を単一の
BEGIN ... COMMITにまとめ、ロードのLabelによって再試行時の重複を排除し、Stream Load の HTTP リクエストのヘッダーから二相コミットを実行できます。最後の仕組みにより、Flink、Spark、CDC パイプラインは独自の管理テーブルを用意せずに、Doris へ厳密に一度だけ書き込めます。
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 によって再試行時の重複を排除し、最後にレプリカへの反映を行うことで、行を一度にクエリ可能にします。
- 開始。 ロードは
PREPARE状態でトランザクションを開始します。FE がTxnIdを割り当て、ユーザーが指定した、または生成された Label に紐付けます。同じ Label で再度送信すると、新しい処理は行われず、元のトランザクションの結果が返されます。 - 書き込み。 BE がデータを受信してセグメントを構築し、成功を報告します。予定されたすべてのロードから確認応答を受けると、トランザクションは
PRECOMMITTEDに進みます。 - コミット。
COMMIT（または第二段階のtxn_operation: commitHTTP 呼び出し）によって、トランザクションはCOMMITTEDに進みます。データはディスク上にありますが、まだ読み取れません。 - 反映。 バックグラウンドのデーモンが各レプリカに
PublishVersionタスクを送ります。影響を受けるすべてのタブレットの全レプリカから確認応答を受けると、トランザクションはVISIBLEになります。 - 読み取り。
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 のロードトランザクションを使う場面
複数の挿入をまとめて反映する必要がある場合、ストリーミングシンクに厳密に一度だけの書き込みが必要な場合、または再試行したロードをアプリケーション側ではなく Label で重複排除する必要がある場合は、Apache Doris のロードトランザクションを使用します。
適しているケース
- チェックポイント間で厳密に一度だけの書き込みを実現するために、Stream Load 2PC を利用する Flink や Spark のシンク。
- ファクトテーブルとディメンションテーブルをまとめて更新する必要がある CDC パイプライン。
- 関連する行のバッチを単位として追加または削除するバックフィル。
- 冪等な再試行が重要で、クライアントではなくデータウェアハウスに Label による重複排除を行わせたいワークロード。
適していないケース
- 高頻度の単一行の挿入。トランザクションごとにコミットの往復通信とタブレットのバージョンが増え、小規模なロードはコミット段階がボトルネックになります。代わりに Group Commit を使用してください。
- 二つのデータベースをまたぐアトミックな処理。単一のトランザクション内のすべてのテーブルは、同じデータベースに属する必要があります。
- クラウド環境またはストレージとコンピュートを分離した構成における、Merge-on-Write の Unique Key テーブルに対する Stream Load 2PC。この組み合わせは BE によって
Status::NotSupportedで拒否されます。オンプレミスのシェアードナッシング構成のクラスターでは使用できます。 - 長期間にわたる重複排除。Label は約 3 日後、またはデータベースごとに 2,000 件に達した時点の、いずれか早い方で削除されます。その後に同じ Label を再利用しても重複排除されません。長期間保持する冪等性キーはアプリケーション側で管理してください。
- 同じトランザクション内で自分の書き込みを読み取る処理。各文は開始時点のスナップショットに対して実行されるため、先行する
INSERTの結果はCOMMITが完了するまで見えません。
関連情報
- Stream Load: 2PC を利用できる同期 HTTP ロード。
- Group Commit: 明示的な
BEGIN ... COMMITを必要としない、高頻度の小規模な挿入に適した方法。 - Kafka と CDC の統合: Routine Load と Flink CDC が、このトランザクションモデルを使って厳密に一度だけの書き込みを実現する仕組み。
- Binlog テーブルストリーム: コミットされた行が可視になったときに生成される変更フィード。
- トランザクション、分離レベル、Label の保持期間
- Stream Load 2PC の手順
- Flink による Doris へのリアルタイム書き込みが高速かつ厳密に一度だけとなる仕組み
- Unique Key: Merge-on-Write による更新も同じロードトランザクションプロトコルでコミットされるテーブルモデル。
