ローカルファイルシステムからデータをロードする
StarRocks はローカルファイルシステムからデータをロードするための2つの方法を提供しています。
- Stream Load を使用した同期ロード
- Broker Load を使用した非同期ロード
それぞれのオプションには独自の利点があります。
- Stream Load は CSV および JSON ファイル形式をサポートします。この方法は、個々のサイズが 10 GB を超えない少数のファイルからデータをロードしたい場合に推奨されます。
- Broker Load は Parquet、ORC、CSV、および JSON ファイル形式をサポートします(JSON ファイル形式は v3.2.3 以降でサポートされています)。この方法は、個々のサイズが 10 GB を超える多数のファイルからデータをロードしたい場合、またはファイルがネットワーク接続ストレージ(NAS)デバイスに保存されている場合に推奨されます。ローカルファイルシステムからデータをロードするために Broker Load を使用することは v2.5 以降でサポートされています。
CSV データについては、以下の点に注意してください。
- カンマ(,)、タブ、またはパイプ(|)などの UTF-8 文字列をテキスト区切り文字として使用できます。長さ は 50 バイトを超えないようにしてください。
- Null 値は
\Nを使用して示されます。たとえば、データファイルが 3 つの列で構成されており、そのデータファイルのレコードが最初と 3 番目の列にデータを保持し、2 番目の列にデータがない場合、この状況では 2 番目の列に\Nを使用して null 値を示す必要があります。つまり、レコードはa,\N,bとしてコンパイルされる必要があります。a,,bは、レコードの 2 番目の列が空の文字列を保持していることを示します。
Stream Load と Broker Load はどちらもデータロード時にデータ変換をサポートし、データロード時に UPSERT および DELETE 操作によるデータ変更をサポートします。詳細については、Transform data at loading および Change data through loading を参照してください。
始める前に
権限の確認
StarRocks のテーブルにデータを ロード するには、その StarRocks テーブルに対して INSERT 権限を持つユーザーである必要があります。INSERT 権限がない場合は、GRANT に記載されている手順に従って、StarRocks クラスター に接続するために使用するユーザーに INSERT 権限を付与してください。構文は GRANT INSERT ON TABLE <table_name> IN DATABASE <database_name> TO { ROLE <role_name> | USER <user_identity>} です。
ネットワーク設定の確認
ロードしたいデータが存在するマシンが、StarRocks クラスタの FE および BE ノードに http_port(デフォルト: 8030)および be_http_port(デフォルト: 8040)を介してアクセスできることを確認してください。
Stream Load を介したローカルファイルシステムからのロード
Stream Load は HTTP PUT ベースの同期ロード方法です。ロードジョブを送信すると、StarRocks はジョブを同期的に実行し、ジョブが終了した後にその結果を返します。ジョブの結果に基づいて、ジョブが成功したかどうかを判断できます。
注意
Stream Load を使用して StarRocks テーブルにデータをロードした後、そのテーブルに作成されたマテリアライズドビューのデータも更新されます。
動作の仕組み
クライアントで HTTP に従って FE にロードリクエストを送信できます。FE は HTTP リダイレクトを使用して特定の BE または CN にロードリクエストを転送します。また、クライアントで選択した BE または CN に直接ロードリクエストを送信することもできます。
FE にロードリクエストを送信する場合、FE はポーリングメカニズムを使用して、どの BE または CN がロードリクエストを受信して処理するコーディネーターとして機能するかを決定します。ポーリングメカニズムは、StarRocks クラスタ内での負荷分散を実現するのに役立ちます。したがって、FE にロードリクエス トを送信することをお勧めします。
ロードリクエストを受信した BE または CN は、コーディネーター BE または CN として機能し、使用されるスキーマに基づいてデータを分割し、データの各部分を他の関連する BE または CN に割り当てます。ロードが終了すると、コーディネーター BE または CN はロードジョブの結果をクライアントに返します。コーディネーター BE または CN をロード中に停止すると、ロードジョブは失敗します。
以下の図は、Stream Load ジョブのワークフローを示しています。

制限
Stream Load は、JSON 形式の列を含む CSV ファイルのデータロードをサポートしていません。
典型的な例
このセクションでは、curl を例にして、ローカルファイルシステムから StarRocks に CSV または JSON ファイルのデータをロードする方法を説明します。詳細な構文とパラメータの説明については、STREAM LOAD を参照してください。
StarRocks では、いくつかのリテラルが SQL 言語によって予約キーワードとして使用されます。これらのキーワードを SQL ステートメ ントで直接使用しないでください。SQL ステートメントでそのようなキーワードを使用したい場合は、バッククォート (`) で囲んでください。Keywords を参照してください。
CSV データのロード
データセットの準備
ローカルファイルシステムで、example1.csv という名前の CSV ファイルを作成します。このファイルは、ユーザー ID、ユーザー名、ユーザースコアを順に表す 3 つの列で構成されています。
1,Lily,23
2,Rose,23
3,Alice,24
4,Julia,25
データベースとテーブルの作成
データベースを作成し、切り替えます。
CREATE DATABASE IF NOT EXISTS mydatabase;
USE mydatabase;
table1 という名前の主キーテーブルを作成します。このテーブルは、id、name、score の 3 つの列で構成されており、id が主キーです。
CREATE TABLE `table1`
(
`id` int(11) NOT NULL COMMENT "user ID",
`name` varchar(65533) NULL COMMENT "user name",
`score` int(11) NOT NULL COMMENT "user score"
)
ENGINE=OLAP
PRIMARY KEY(`id`)
DISTRIBUTED BY HASH(`id`);
v2.5.7 以降、StarRocks はテーブルを作成する際やパーティションを追加する際に、バケット数 (BUCKETS) を自動的に設定できます。バケット数を手動で設定する必要はありません。詳細については、set the number of buckets を参照してください。
Stream Load の開始
次のコマンドを実行して、example1.csv のデータを table1 にロードします。
curl --location-trusted -u <username>:<password> -H "label:123" \
-H "Expect:100-continue" \
-H "column_separator:," \
-H "columns: id, name, score" \
-T example1.csv -XPUT \
http://<fe_host>:<fe_http_port>/api/mydatabase/table1/_stream_load
- パスワードが設定されていないアカウントを使用する場合は、
<username>:のみを入力する必要があります。 - SHOW FRONTENDS を使用して、FE ノードの IP アドレスと HTTP ポートを表示できます。
example1.csv は 3 つの列で構成されており、カンマ (,) で区切られ、table1 の id、name、score 列に順番にマッピングされます 。したがって、column_separator パラメータを使用してカンマ (,) を列区切り文字として指定する必要があります。また、columns パラメータを使用して、example1.csv の 3 つの列を一時的に id、name、score として名前を付け、table1 の 3 つの列に順番にマッピングする必要があります。
ロードが完了したら、table1 をクエリしてロードが成功したことを確認できます。
SELECT * FROM table1;
+------+-------+-------+
| id | name | score |
+------+-------+-------+
| 1 | Lily | 23 |
| 2 | Rose | 23 |
| 3 | Alice | 24 |
| 4 | Julia | 25 |
+------+-------+-------+
4 rows in set (0.00 sec)
JSON データ のロード
v3.2.7 以降、Stream Load は JSON データの圧縮をサポートしており、ネットワーク帯域幅のオーバーヘッドを削減します。ユーザーは compression および Content-Encoding パラメータを使用して異なる圧縮アルゴリズムを指定できます。サポートされている圧縮アルゴリズムには GZIP、BZIP2、LZ4_FRAME、ZSTD があります。構文については、STREAM LOAD を参照してください。
データセットの準備
ローカルファイルシステムで、example2.json という名前の JSON ファイルを作成します。このファイルは、都市 ID と都市名を順に表す 2 つの列で構成されています。
{"name": "Beijing", "code": 2}
データベースとテーブルの作成
データベースを作成し、切り替えます。
CREATE DATABASE IF NOT EXISTS mydatabase;
USE mydatabase;
table2 という名前の主キーテーブルを作成します。このテーブルは、id と city の 2 つの列で構成されており、id が主キーです。
CREATE TABLE `table2`
(
`id` int(11) NOT NULL COMMENT "city ID",
`city` varchar(65533) NULL COMMENT "city name"
)
ENGINE=OLAP
PRIMARY KEY(`id`)
DISTRIBUTED BY HASH(`id`);
v2.5.7 以降、StarRocks はテーブルを作成する際やパーティションを追加する際に、バケット数 (BUCKETS) を自動的に設定できます。バケット数を手動で設定する必要はありません。詳細については、set the number of buckets を参照してください。
Stream Load の開始
次のコマンドを実行して、example2.json のデータを table2 にロードします。
curl -v --location-trusted -u <username>:<password> -H "strict_mode: true" \
-H "Expect:100-continue" \
-H "format: json" -H "jsonpaths: [\"$.name\", \"$.code\"]" \
-H "columns: city,tmp_id, id = tmp_id * 100" \
-T example2.json -XPUT \
http://<fe_host>:<fe_http_port>/api/mydatabase/table2/_stream_load
- パスワードが設定されていないアカウントを使用する場合は、
<username>:のみを入力する必要があります。 - SHOW FRONTENDS を使用して、FE ノードの IP アドレスと HTTP ポートを表示できます。
example2.json は name と code の 2 つのキーで構成されており、table2 の id と city 列にマッピングされます。以下の図に示されています。

上記の図に示されているマッピングは次のように説明されます。
-
StarRocks は
example2.jsonのnameとcodeキーを抽出し、それらをjsonpathsパラメータで宣言されたnameとcodeフィールドにマッピングします。 -
StarRocks は
jsonpathsパラメータで宣言されたnameとcodeフィールドを抽出し、それらをcolumnsパラメータで宣言されたcityとtmp_idフィールドに順番にマッピングします。 -
StarRocks は
columnsパラメータで宣言されたcityとtmp_idフィールドを抽出し、それらをtable2のcityとid列に名前でマッピングします。
上記の例では、example2.json の code の値は table2 の id 列にロードされる前に 100 倍されます。
StarRocks テーブルの jsonpaths、columns、および列の詳細なマッピングについては、STREAM LOAD の「Column mappings」セクションを参照してください。
ロードが完了したら、table2 をクエリしてロードが成功したことを確認できます。
SELECT * FROM table2;
+------+--------+
| id | city |
+------+--------+
| 200 | Beijing|
+------+--------+
4 rows in set (0.01 sec)
Stream Load リクエストのマージ
v3.4.0 以降、システムは複数の Stream Load リクエストのマージをサポートしています。
Merge Commit 最適化は、単一テーブルに対する並行 Stream Load ジョブが実行されるシナリオに適しています。並行度が 1 の場合は推奨されません。また、merge_commit_async を false に設定し、merge_commit_interval_ms を大きな値に設定する場合は慎重に検討してください。これらはロードパフォーマンスの低下を引き起こす可能性があります。
Merge Commit は、hc、小バッチ(KB から数十 MB)のリアルタイムロードシナリオ向けに設計された Stream Load の最適化です。以前のバージョンでは、各 Stream Load リクエストがトランザクションとデータバージョンを生成していましたが、hc ロードシナリオでは次のような問題が発生していました。
- データバージョンが多すぎるとクエリパフォーマンスに影響し、バージョン数を制限すると
too many versionsエラーが発生する可能性があります。 - Compaction を通じたデータバージョンのマージはリソース消費を増加させます。
- 小さなファイルが生成され、IOPS と I/O レイテンシが増加します。共有データクラスタでは、クラウドオブジェクトストレージのコストも増加します。
- トランザクションマネージャーとしての Leader FE ノードがボトルネックとなる可能性があります。
Merge Commit は、時間ウィンドウ内で複数の同時 Stream Load リクエストを単一のトランザクションにマージすることで、これらの問題を軽減します。これにより、hc リクエストによって生成されるトランザクションとバージョンの数が減少し、ロードパフォーマンスが向上します。
Merge Commit は同期モードと非同期モードの両方をサポートしています。各モードには利点と欠点があります。使用ケースに基づいて選択できます。
-
同期モード
サーバーは、マージされたトランザクションがコミットされた後にのみ返します。これにより、ロードが成功し、可視性が保証されます。
-
非同期モード
サーバーはデータを受信した直後に返します。このモードでは、ロードが成功することを保証しません。
| モード | 利点 | 欠点 |
|---|---|---|
| 同期モード |
| クライアントからの各ロードリクエストは、サーバーがマージウィンドウを閉じるまでブロックされます。ウィンドウが過度に大きい場合、単一クライアントのデータ処理能力が低下する可能性があります。 |
| 非同期モード | 単一のクライアントがサーバーがマージウィンドウを閉じるのを待たずに後続のロードリクエストを送信でき、ロードスループットが向上します。 |
|
Stream Load の開始
-
次のコマンドを実行して、同期モードで Merge Commit が有効になっている Stream Load ジョブを開始し、マージウィンドウを
5000ミリ秒に、並行度を2に設定します。curl --location-trusted -u <username>:<password> \
-H "Expect:100-continue" \
-H "column_separator:," \
-H "columns: id, name, score" \
-H "enable_merge_commit:true" \
-H "merge_commit_interval_ms:5000" \
-H "merge_commit_parallel:2" \
-T example1.csv -XPUT \
http://<fe_host>:<fe_http_port>/api/mydatabase/table1/_stream_load -
次のコマンドを実行して、非同期モードで Merge Commit が有効になっている Stream Load ジョブを開始し、マージ ウィンドウを
60000ミリ秒に、並行度を2に設定します。curl --location-trusted -u <username>:<password> \
-H "Expect:100-continue" \
-H "column_separator:," \
-H "columns: id, name, score" \
-H "enable_merge_commit:true" \
-H "merge_commit_async:true" \
-H "merge_commit_interval_ms:60000" \
-H "merge_commit_parallel:2" \
-T example1.csv -XPUT \
http://<fe_host>:<fe_http_port>/api/mydatabase/table1/_stream_load
- Merge Commit は、同質の ロードリクエストを単一のデータベースとテーブルにマージすることのみをサポートしています。 「同質」とは、Stream Load パラメータが共通パラメータ、JSON 形式のパラメータ、CSV 形式のパラメータ、
opt_properties、および Merge Commit パラメータを含めて同一であることを意味します。 - CSV 形式のデータをロードする場合、各行が行区切り文字で終わることを確認する必要があります。
skip_headerはサポートされていません。 - サーバーはトランザクションのラベルを自動的に生成します。指定された場合は無視されます。
- Merge Commit は、複数のロードリクエストを単一のトランザクションにマージします。1 つのリクエストにデータ品質の問題が含まれている場合、そのトランザクション内のすべてのリクエストが失敗します。
Stream Load の進捗確認
ロードジョブが完了すると、StarRocks はジョブの結果を JSON 形式で返します。詳細については、STREAM LOAD の「Return value」セクションを参照してください。
Stream Load では、SHOW LOAD ステートメントを使用してロードジョブの結果をクエリすることはできません。