Stream Load HTTP PUT を介した Doris へのリアルタイムデータ取り込み - E-MapReduce

Stream Load は、ローカルファイルまたはデータストリームを Apache Doris にインポートするための、HTTP ベースの同期メソッドです。HTTP PUT リクエストを介してデータを送信すると、インポート結果がすぐに取得されるため、ポーリングは不要です。

サポートされているデータ形式：CSV および JSON。

仕組み

インポートジョブを送信すると、次のようになります：

リクエストをフロントエンド (FE) に送信すると、FE は HTTP リダイレクトを介してコーディネーターバックエンド (BE) にリダイレクトします。
リクエストを直接 BE に送信すると、そのノードがコーディネーター BE として機能します。
コーディネーター BE はデータを受信し、他の BE に配布します。
インポートが完了すると、コーディネーター BE は HTTP 応答で結果を返します。

重要

Stream Load は同期的です。インポート結果は Doris に記録されないため、SHOW LOAD では Stream Load ジョブは表示されません。成功か失敗かを判断するには、HTTP 応答を直接確認してください。

クイックスタート

次の例では、ローカルの CSV ファイルを Doris テーブルにインポートします。プレースホルダーを置き換えて、コマンドを直接実行してください。

curl --location-trusted \
  -u <user>:<passwd> \
  -H "label:my-first-load" \
  -H "column_separator:," \
  -T /path/to/data.csv \
  -XPUT http://<fe-host>:<http-port>/api/<db>/<table>/_stream_load

成功した応答は次のようになります：

{
    "TxnId": 1003,
    "Label": "my-first-load",
    "Status": "Success",
    "Message": "OK",
    "NumberTotalRows": 1000000,
    "NumberLoadedRows": 1000000,
    "NumberFilteredRows": 1,
    "NumberUnselectedRows": 0,
    "LoadBytes": 40888898,
    "LoadTimeMs": 2144,
    "BeginTxnTimeMs": 1,
    "StreamLoadPutTimeMs": 2,
    "ReadDataTimeMs": 325,
    "WriteDataTimeMs": 1933,
    "CommitAndPublishTimeMs": 106,
    "ErrorURL": "http://192.168.x.x:8042/api/_load_error_log?file=__shard_0/error_log_insert_stmt_****"
}

インポートジョブの送信

前提条件

開始する前に、次のものが揃っていることを確認してください：

少なくとも 1 つの FE と 1 つの BE を持つ実行中の Doris クラスター
FE または BE ノードのホストと HTTP ポート
Doris で既に作成されているデータベースとテーブル
ターゲットデータベースに対するインポート権限を持つ認証情報

コマンド構文

curl --location-trusted -u <user>:<passwd> \
  [-H "<key>:<value>" ...] \
  -T <data-file> \
  -XPUT http://<fe-host>:<http-port>/api/<db>/<table>/_stream_load

すべてのインポートパラメーターは、-H "key:value" を使用して HTTP ヘッダーとして渡します。完全な構文を表示するには、Doris で HELP STREAM LOAD を実行してください。

リクエストパラメーター

認証

パラメーター	説明
`user:passwd`	Basic 認証のための認証情報。Doris は、これらの認証情報に基づいて ID とインポート権限を検証します。

ジョブ制御

パラメーター	デフォルト	説明
`label`	自動生成	データベース内でインポートジョブを一意に識別する ID。同じバッチのデータに同じラベルを使用すると、高々 1 回のセマンティクスが適用されます。つまり、既存のラベルを持つ重複リクエストは 1 回だけ受け入れられます。`CANCELLED` 状態のラベルは再利用できます。
`max_filter_ratio`	`0`	ジョブが失敗する前に許容される最大フィルター率。有効な値：`0`～`1`。`0` より大きい値を設定すると、不正な行をスキップできます。計算式：`dpp.abnorm.ALL / (dpp.abnorm.ALL + dpp.norm.ALL) > max_filter_ratio`。
`where`	—	ソース行をフィルターするための WHERE 句。フィルターされた行は `NumberUnselectedRows` でカウントされ、エラー率の計算には含まれません。
`exec_mem_limit`	`2147483648` (2 GB)	インポートジョブのメモリ制限 (バイト単位)。
`strict_mode`	無効	`true` に設定すると、Strict モードが有効になります。Strict モードでは、ソースで列の値が非 null であっても、列の型変換後に null に変換される行はフィルターされます。Strict モードは、null 値が関数評価に由来する列や、変換された値がターゲット型の範囲外である列 (例：`10` 列に対するソース値 `DECIMAL(1,0)`) には適用されません。

データマッピング

パラメーター	デフォルト	説明
`column_separator`	`\t`	列区切り文字。印刷不可文字の場合は、`\x` プレフィックスを付けた 16 進数形式を使用します (例：Hive ファイルの場合は `\x01`)。複数文字の区切り文字もサポートされています。
`line_delimiter`	`\n`	行区切り文字。複数文字の区切り文字もサポートされています。
`columns`	—	列マッピングと式変換。列の並べ替えと、クエリ文と同じ式構文を使用した関数ベースの変換をサポートします。
`Partitions`	—	ターゲットパーティション。指定されたパーティションに属さない行は除外されます。除外された行は `dpp.abnorm.ALL` でカウントされます。

データマージ

パラメーター	デフォルト	値
`merge_type`	`APPEND`	`APPEND`：このバッチを既存のデータに追加します。`DELETE`：このバッチとキーが一致する行を削除します。`MERGE`：DELETE 条件に一致する行には `DELETE` セマンティクスを適用し、それ以外の行には `APPEND` セマンティクスを適用します。

2 フェーズコミット

2 フェーズコミットを使用すると、データの書き込みと可視化を 2 つの別々のステップで行うことができます。データは書き込まれますが、手動でコミットをトリガーするまで非表示のままです (トランザクション状態：PRECOMMITTED)。

2 フェーズコミットを有効にするには：

be.conf で disable_stream_load_2pc=false を設定します。
リクエストヘッダーで two_phase_commit:true を設定します。

ステップ 1：2 フェーズコミットを有効にしてインポートジョブを送信

curl --location-trusted -u <user>:<passwd> \
  -H "two_phase_commit:true" \
  -T test.txt \
  http://<fe-host>:<http-port>/api/<db>/<table>/_stream_load

応答にはトランザクション ID (TxnId) が含まれ、"TwoPhaseCommit": "true" が確認されます。データは書き込まれましたが、まだ可視化されていません。

ステップ 2：トランザクションのコミットまたは中止

応答の TxnId を使用して、コミット操作 (データを可視化) または中止操作 (書き込まれたデータを破棄) を手動でトリガーします。

インポート結果の表示

戻り値フィールド

フィールド	説明
`TxnId`	インポートジョブのトランザクション ID。トランザクション ID は Alibaba Cloud によってフルマネージドにできます。
`Label`	インポートジョブのラベル。
`Status`	`Success`：インポート成功。`Publish Timeout`：インポートは完了しましたが、データの可視化が遅れる可能性があります。リトライは不要です。`Label Already Exists`：ラベルを変更してリトライしてください。`Fail`：インポート失敗。
`ExistingJobStatus`	既存のラベルに関連付けられたジョブのステータス。`Status` が `Label Already Exists` の場合にのみ表示されます。値：`RUNNING` または `FINISHED`。
`Message`	エラーメッセージ (ある場合)。
`NumberTotalRows`	処理された総行数。
`NumberLoadedRows`	行が正常にインポートされました。
`NumberFilteredRows`	インポートに失敗した行数。
`NumberUnselectedRows`	WHERE 句によってフィルターされた行。
`LoadBytes`	インポートされたバイト数。
`LoadTimeMs`	総インポート時間 (ミリ秒単位)。
`BeginTxnTimeMs`	FE トランザクションの開始時間 (ミリ秒単位)。
`StreamLoadPutTimeMs`	FE 実行計画の取得時間 (ミリ秒単位)。
`ReadDataTimeMs`	データ読み取り時間 (ミリ秒単位)。
`WriteDataTimeMs`	データ書き込み時間 (ミリ秒単位)。
`CommitAndPublishTimeMs`	FE がトランザクションをコミットして公開するまでの時間 (ミリ秒単位)。
`ErrorURL`	インポートに失敗した行を調査するための URL。

過去のインポートジョブの表示

完了したインポートジョブをクエリするには、SHOW STREAM LOAD を実行します。

デフォルトでは、BE は Stream Load ジョブの履歴を記録しません。記録を有効にするには、BE 設定で enable_stream_load_record を true に設定します。詳細については、「バックエンドノードの設定項目」をご参照ください。

インポートジョブのキャンセル

Stream Load ジョブは手動でキャンセルできません。タイムアウトまたはインポートエラーが発生した場合、システムはジョブを自動的にキャンセルします。

システム設定

Stream Load の動作は 2 つのレベルで制御されます。運用チームは、FE および BE 設定ファイルでクラスター全体のデフォルト値を設定します。個々のジョブでは、HTTP リクエストヘッダーで個別のタイムアウト期間を設定することで、ジョブごとのタイムアウトをオーバーライドできます。

FE 設定

パラメーター	デフォルト	説明
`stream_load_default_timeout_second`	`600`	インポートジョブのグローバルなデフォルトタイムアウト (秒単位)。ジョブがこの期間内に完了しない場合、システムはジョブをキャンセルし、ステータスを `CANCELLED` に設定します。HTTP リクエストヘッダーで個別のタイムアウト期間を設定してジョブごとにオーバーライドするか、このパラメーターを更新してクラスター全体のデフォルト値を変更します。

BE 設定

パラメーター	デフォルト	説明
`streaming_load_max_mb`	`10240` (10 GB)	インポートジョブごとの最大データサイズ (MB 単位)。ソースファイルがこの制限を超える場合は、ジョブを送信する前にこのパラメーターを増やしてください。

ベストプラクティス

Stream Load の使用時期

Stream Load は、ソースファイルがすでにメモリまたはローカルディスクにあり、インポート結果を同期的に取得する必要がある場合に最適です。

推奨バッチサイズ

ジョブごとに 1 GB から 10 GB をインポートします。デフォルトの最大値は 10 GB (streaming_load_max_mb = 10240) です。より大きなファイルの場合は、送信前に be.conf の streaming_load_max_mb を増やしてください。

同時 Stream Load ジョブはクラスターサイズによって制限されません。

タイムアウト計画

次の数式を使用して、必要なタイムアウトを見積もります：

タイムアウト (秒) = ファイルサイズ (MB) / 10 MB/秒

実際のスループットはクラスターによって異なります。観測されたパフォーマンスに基づいて数式を調整してください。

見積もられたタイムアウトがデフォルトの 600 秒を超える場合は、HTTP リクエストヘッダーでタイムアウト期間を指定してジョブごとに設定するか、fe.conf の stream_load_default_timeout_second を介してグローバルに、より長い値を設定してください。

完全な例

シナリオ：ローカルディスクの /home/store_sales から 15 GB のファイルを bj_sales データベースの store_sales テーブルにインポートします。

ステップ 1：BE のサイズ制限を増やす

ファイルはデフォルトの 10 GB を超えています。be.conf に次の行を追加し、BE を再起動します：

streaming_load_max_mb = 16000

ステップ 2：必要なタイムアウトを計算する

15,000 MB / 10 MB/秒 = 1,500 秒

1,500 秒はデフォルトの 600 秒を超えるため、fe.conf を更新します：

stream_load_default_timeout_second = 1500

ステップ 3：インポートジョブを送信する

curl --location-trusted \
  -u user:password \
  -H "label:abc" \
  -T /home/store_sales \
  -XPUT http://abc.com:8030/api/bj_sales/store_sales/_stream_load