E-MapReduce:Stream Load

用語

coordinator: データを受信し、他のデータノードにデータを配信し、データのインポート後に結果を返すノード。

基本原則

Stream Load では、HTTP リクエストを送信することでインポートジョブを送信できます。フロントエンド（FE）ノードにリクエストを送信すると、FE ノードは HTTP リダイレクトを実行してリクエストをバックエンド（BE）ノードに転送します。BE ノードに直接リクエストを送信することもできます。 BE ノードはコーディネーターノードとして機能し、テーブルスキーマによってデータを分割し、関連する BE ノードにデータを配信します。その後、コーディネーターノードはインポートジョブの結果を返します。

サポートされているヘッダープロパティは、次の表のインポートジョブパラメーターの説明で確認できます。パラメーターは -H "key1:value1" 形式で構成されます。複数のパラメーターが関係する場合は、複数 -H を使用してパラメーターを示す必要があります。例: -H "key1:value1" -H "key2:value2"。 Stream Load モードでは、インポートジョブに関連するすべてのパラメーターがヘッダーで構成されます。次の表にパラメーターを示します。

ベストプラクティス

シナリオ

Stream Load は、ソースファイルがメモリまたはローカルディスクに保存されているシナリオに適しています。 Stream Load は同期インポート方式です。インポートジョブの結果を同期的に取得する場合は、Stream Load でインポートジョブを送信できます。

データサイズ

Stream Load モードでは、BE ノードがデータをインポートおよび配信します。 1 GB ～ 10 GB のデータをインポートするには、Stream Load を使用することをお勧めします。デフォルトでは、Stream Load モードでインポートできるデータの最大サイズは 10 GB です。したがって、サイズが 10 GB を超えるファイルをインポートするには、BE ノードの streaming_load_max_mb パラメーターを変更する必要があります。たとえば、インポートするファイルのサイズが 15 GB（15,360 MB）の場合、BE ノードの streaming_load_max_mb パラメーターを 15 GB を超えるデータサイズに設定できます。

curl --location-trusted -u 'admin:****'   -XPOST http://be-c-****-internal.starrocks.aliyuncs.com:8040/api/update_config?streaming_load_max_mb=15360

Stream Load で送信されるインポートジョブのデフォルトのタイムアウト期間は 600 秒です。タイムアウト期間を調整するには、EMR コンソール で、StarRocks インスタンスの FE ノードの timeout パラメーターの値を変更します。

完全な例

データは、特定のクライアントの /mnt/disk1/customer.tbl ディレクトリに保存されています。データを StarRocks インスタンスの stream_load データベースの customer テーブルにインポートします。

データのダウンロード: customer.tbl

インスタンス情報: Stream Load モードで同時に処理できるインポートジョブの数は、インスタンスサイズの影響を受けません。

手順:

1. インポートするファイルのサイズがデフォルトの上限を超えている場合は、BE ノードの BE.conf 構成ファイルを変更します。たとえば、streaming_load_max_mb パラメーターを 15360 に設定できます。単位: MB。

   curl --location-trusted -u 'admin:*****'  -XPOST http://be-c-****-internal.starrocks.aliyuncs.com:8040/api/update_config?streaming_load_max_mb=15360

2. インスタンスの [インスタンス構成] タブで、stream_load_default_timeout_second パラメーターの値を変更します。この例では、値を 3600 に設定します。

3. customer という名前の宛先テーブルを作成します。

   CREATE TABLE `customer` (
     `c_custkey` bigint(20) NULL COMMENT "",
     `c_name` varchar(65533) NULL COMMENT "",
     `c_address` varchar(65533) NULL COMMENT "",
     `c_nationkey` bigint(20) NULL COMMENT "",
     `c_phone` varchar(65533) NULL COMMENT "",
     `c_acctbal` double NULL COMMENT "",
     `c_mktsegment` varchar(65533) NULL COMMENT "",
     `c_comment` varchar(65533) NULL COMMENT ""
   ) ENGINE=OLAP
   DUPLICATE KEY(`c_custkey`)
   COMMENT "OLAP"
   DISTRIBUTED BY HASH(`c_custkey`) BUCKETS 24
   PROPERTIES (
   "replication_num" = "1",
   "in_memory" = "false",
   "storage_format" = "DEFAULT",
   "enable_persistent_index" = "false",
   "compression" = "LZ4"
   );

4. インポートジョブを作成します。データセットに大量のデータが含まれている場合は、バックグラウンドで操作を実行できます。

   curl --location-trusted -u 'admin:*****' -T /mnt/disk1/customer.tbl -H "label:labelname" -H "column_separator:|"  http://fe-c-****-internal.starrocks.aliyuncs.com:8030/api/load_test/customer/_stream_load

   次の情報が返されます。

   {
       "TxnId": 575,
       "Label": "labelname",
       "Status": "Success",
       "Message": "OK",
       "NumberTotalRows": 150000,
       "NumberLoadedRows": 150000,
       "NumberFilteredRows": 0,
       "NumberUnselectedRows": 0,
       "LoadBytes": 24196144,
       "LoadTimeMs": 1081,
       "BeginTxnTimeMs": 104,
       "StreamLoadPlanTimeMs": 106,
       "ReadDataTimeMs": 85,
       "WriteDataTimeMs": 850,
       "CommitAndPublishTimeMs": 20
   }

   説明

   エラー "ErrorURL": "http://***:8040/api/_load_error_log?file=error_log_***" が報告された場合は、curl コマンドを実行して詳細を表示できます。

統合のサンプルコード

• Java を使用して Stream Load でインポートジョブを送信する方法については、「stream_load」をご参照ください。

• Spark を使用して Stream Load でインポートジョブを送信する方法については、「01_sparkStreaming2StarRocks」をご参照ください。