E-MapReduce:Broker Load

ブローカーロードの仕組み

ブローカーロードには、ご利用の StarRocks バージョンに応じて 2 つの動作モードがあります。

• ブローカーベースのロード (StarRocks 2.5.8 より前のバージョン): StarRocks は、ブローカープロセスに依存して外部ストレージに接続します。ロードステートメントで WITH BROKER "<broker_name>" を使用して、ブローカー名を明示的に指定します。

• ブローカーフリーのロード (StarRocks 2.5.8 以降): StarRocks は外部ストレージに直接接続します。ブローカー名は省略しますが、ロードステートメントの WITH BROKER キーワードは保持します。

E-MapReduce (EMR) で StarRocks クラスターを作成すると、ブローカープロセスはすべてのコアノードに自動的にデプロイされ、起動します。

サポートされているソースとフォーマット

前提条件

開始する前に、以下が揃っていることを確認してください。

• 少なくとも 1 つのコアノードを持つ EMR で実行中の StarRocks クラスター

• HDFS または OSS のアクセス認証情報

• StarRocks のターゲットデータベースとテーブル

ブローカーのクエリ

クラスターでブローカーが実行されていることを確認するには、以下を実行します。

SHOW PROC "/brokers"\G

出力には、各ブローカーの IP アドレス、ポート (デフォルト: 8000)、稼働状況、タイムスタンプが一覧表示されます。

*************************** 1. row ***************************
          Name: broker
            IP: 10.0.**.**
          Port: 8000
         Alive: true
 LastStartTime: 2022-04-13 11:38:46
LastUpdateTime: 2022-04-13 15:26:44
        ErrMsg:

インポートジョブの作成

すべてのインポートジョブは LOAD LABEL ステートメントを使用します。構文は StarRocks のバージョンによって若干異なります。

LOAD LABEL db_name.label_name
    (data_desc, ...)
WITH BROKER broker_name broker_properties
    [PROPERTIES (key1=value1, ... )]

LOAD LABEL db_name.label_name
    (data_desc, ...)
WITH BROKER broker_properties
    [PROPERTIES (key1=value1, ... )]

EMR StarRocks クラスターでは、ブローカー名として broker を使用します。

完全な構文リファレンスを表示するには、HELP BROKER LOAD を実行します。

data_desc:
    DATA INFILE ('file_path', ...)
    [NEGATIVE]
    INTO TABLE tbl_name
    [PARTITION (p1, p2)]
    [COLUMNS TERMINATED BY column_separator]
    [FORMAT AS file_type]
    [(col1, ...)]
    [COLUMNS FROM PATH AS (colx, ...)]
    [SET (k1=f1(xx), k2=f2(xx))]
    [WHERE predicate]

broker_properties:
    (key2=value2, ...)

OSS からのデータロード

次の例では、OSS から TPC-H lineitem ファイルを StarRocks テーブルにロードします。

LOAD LABEL tpch.lineitem
(
    DATA INFILE("oss://bucket/tpc_h/sf1/lineitem.tbl")
    INTO TABLE `lineitem`
    COLUMNS TERMINATED BY '|'
    (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice,
     l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate, l_commitdate,
     l_receiptdate, l_shipinstruct, l_shipmode, l_comment)
)

LOAD LABEL tpch.lineitem
(
    DATA INFILE("oss://bucket/tpc_h/sf1/lineitem.tbl")
    INTO TABLE `lineitem`
    COLUMNS TERMINATED BY '|'
    (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice,
     l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate, l_commitdate,
     l_receiptdate, l_shipinstruct, l_shipmode, l_comment)
)
WITH BROKER broker
(
    "fs.oss.accessKeyId" = "xxx",
    "fs.oss.accessKeySecret" = "xxx",
    "fs.oss.endpoint" = "oss-cn-beijing-internal.aliyuncs.com"
);

HDFS からのデータロード

(
    "username" = "user",
    "password" = "passwd",
    "dfs.nameservices" = "my-ha",
    "dfs.ha.namenodes.my-ha" = "my-namenode1,my-namenode2",
    "dfs.namenode.rpc-address.my-ha.my-namenode1" = "nn1-host:rpc_port",
    "dfs.namenode.rpc-address.my-ha.my-namenode2" = "nn2-host:rpc_port",
    "dfs.client.failover.proxy.provider.my-ha" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
)

LOAD LABEL db1.label1
(
    DATA INFILE("hdfs://emr-header-1.cluster-xxx:9000/user/hive/test.db/ml/file1")
    INTO TABLE tbl1
    COLUMNS TERMINATED BY ","
    (tmp_c1, tmp_c2)
    SET
    (
        id = tmp_c2,
        name = tmp_c1
    ),

    DATA INFILE("hdfs://emr-header-1.cluster-xxx:9000/user/hive/test.db/ml/file2")
    INTO TABLE tbl2
    COLUMNS TERMINATED BY ","
    (col1, col2)
    WHERE col1 > 1
)
WITH BROKER 'broker1'
(
    "username" = "hdfs_username",
    "password" = "hdfs_password"
)
PROPERTIES
(
    "timeout" = "3600"
);

ジョブステータスのクエリ

ブローカーロードは非同期です。ジョブを送信した後、SHOW LOAD を使用して進行状況を追跡します。

SHOW LOAD WHERE label = 'label1'\G

完全な SHOW LOAD 構文を表示するには、HELP SHOW LOAD を実行します。

SHOW LOAD は非同期インポートメソッドでのみ機能します。Stream Load のような同期メソッドには適用されません。

出力は次のようになります。

*************************** 1. row ***************************
         JobId: 7****
         Label: label1
         State: FINISHED
      Progress: ETL:N/A; LOAD:100%
          Type: BROKER
       EtlInfo: unselected.rows=4; dpp.abnorm.ALL=15; dpp.norm.ALL=28133376
      TaskInfo: cluster:N/A; timeout(s):10800; max_filter_ratio:5.0E-5
      ErrorMsg: N/A
    CreateTime: 2019-07-27 11:46:42
  EtlStartTime: 2019-07-27 11:46:44
 EtlFinishTime: 2019-07-27 11:46:44
 LoadStartTime: 2019-07-27 11:46:44
LoadFinishTime: 2019-07-27 11:50:16
           URL: http://192.168.**.**:8040/api/_load_error_log?file=__shard_4/error_log_insert_stmt_...
    JobDetails: {"Unfinished backends":{"9c3441027ff948a0-8287923329a2****":[10002]},"ScannedRows":2390016,"TaskNumber":1,"All backends":{"9c3441027ff948a0-8287923329a2****":[10002]},"FileNumber":1,"FileSize":1073741824}

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

PENDING または LOADING 状態のジョブをキャンセルします。

CANCEL LOAD
[FROM db_name]
WHERE [LABEL = "load_label" | LABEL LIKE "label_pattern"];

完全な構文を表示するには、HELP CANCEL LOAD を実行します。

エンドツーエンドの例: OSS からのロード

この例では、OSS から StarRocks に TPC-H lineitem データをロードします。

ステップ 1: ターゲットテーブルを作成します。

CREATE TABLE lineitem (
  l_orderkey      bigint,
  l_partkey       bigint,
  l_suppkey       bigint,
  l_linenumber    int,
  l_quantity      double,
  l_extendedprice double,
  l_discount      double,
  l_tax           double,
  l_returnflag    string,
  l_linestatus    string,
  l_shipdate      date,
  l_commitdate    date,
  l_receiptdate   date,
  l_shipinstruct  string,
  l_shipmode      string,
  l_comment       string
)
ENGINE = OLAP
DUPLICATE KEY(l_orderkey)
DISTRIBUTED BY HASH(l_orderkey) BUCKETS 96
PROPERTIES (
  "replication_num" = "1"
);

ステップ 2: インポートジョブを送信します。

LOAD LABEL tpch.lineitem
(
    DATA INFILE("oss://xxx/tpc_h/sf1/lineitem.tbl")
    INTO TABLE `lineitem`
    COLUMNS TERMINATED BY '|'
    (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice,
     l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate, l_commitdate,
     l_receiptdate, l_shipinstruct, l_shipmode, l_comment)
)

ステップ 3: ジョブを監視します。

SHOW LOAD WHERE label = 'lineitem'\G

ジョブが成功すると、出力に State: FINISHED と Progress: ETL:100%; LOAD:100% が表示されます。

ステップ 4: データを検証します。

-- Check total row count
SELECT COUNT(*) FROM lineitem;

-- Preview the first two rows
SELECT * FROM lineitem LIMIT 2;

同時実行性

ブローカーロードジョブはタスクに分割され、さらにタスクはバックエンド全体に並行して分散されるインスタンスに分割されます。

タスク分割: ジョブは、異なるソースアドレスまたは異なるパーティションセットを指す data_desc 句ごとに 1 つのタスクを生成します。

インスタンス分割: 各タスクは、次の 3 つのフロントエンドパラメーターに基づいてインスタンスに分割されます。

総インスタンス数 = min(ファイルサイズ / min_bytes_per_broker_scanner, max_broker_concurrency, load_parallel_instance_num × バックエンド数)

実際には、ほとんどのジョブには単一の data_desc 句があり、したがって単一のタスクがあります。そのタスクは、バックエンドの数と同じ数のインスタンスに分割され、各インスタンスは異なるバックエンドで実行されます。

トラブルシューティング

次のステップ

• 高度な LOAD LABEL オプションと HELP BROKER LOAD の使用法については、StarRocks ドキュメントをクエリしてください。

• SHOW LOAD で進行中のジョブを監視し、パフォーマンスチューニングのためにバックエンドログを確認してください。

• クラスターのインポート同時実行数を調整するために、フロントエンドで min_bytes_per_broker_scanner、max_broker_concurrency、および load_parallel_instance_num を調整してください。