E-MapReduce:Broker Load

Cara kerja Broker Load

Broker Load memiliki dua mode operasi tergantung pada versi StarRocks yang Anda gunakan:

• Pemuatan berbasis broker (StarRocks versi sebelum 2.5.8): StarRocks bergantung pada proses broker untuk terhubung ke penyimpanan eksternal. Tentukan nama broker secara eksplisit dengan WITH BROKER "<broker_name>" dalam pernyataan pemuatan.

• Pemuatan tanpa broker (StarRocks 2.5.8 dan lebih baru): StarRocks terhubung langsung ke penyimpanan eksternal. Abaikan nama broker tetapi tetap gunakan kata kunci WITH BROKER dalam pernyataan pemuatan.

Saat Anda membuat kluster StarRocks di E-MapReduce (EMR), proses broker secara otomatis dideploy dan dijalankan di setiap node inti.

Sumber dan format yang didukung

Prasyarat

Sebelum memulai, pastikan Anda telah memiliki:

• Kluster StarRocks yang sedang berjalan di EMR dengan minimal satu node inti

• Kredensial akses untuk HDFS atau OSS

• Database dan tabel tujuan di StarRocks

Kueri Broker

Untuk memastikan broker sedang berjalan di kluster Anda, jalankan:

SHOW PROC "/brokers"\G

Output akan menampilkan daftar setiap broker beserta alamat IP-nya, port (default: 8000), status alive, dan timestamp:

*************************** 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:

Membuat pekerjaan impor

Semua pekerjaan impor menggunakan pernyataan LOAD LABEL. Sintaksnya sedikit berbeda tergantung versi 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, ... )]

Di kluster StarRocks EMR, gunakan broker sebagai nama broker.

Untuk melihat referensi sintaks lengkap, jalankan 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, ...)

Memuat data dari OSS

Contoh berikut memuat file lineitem TPC-H dari OSS ke tabel 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"
);

Memuat data dari 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"
);

Memeriksa status pekerjaan

Broker Load bersifat asinkron. Setelah mengirim pekerjaan, gunakan SHOW LOAD untuk melacak progresnya:

SHOW LOAD WHERE label = 'label1'\G

Untuk melihat sintaks lengkap SHOW LOAD, jalankan HELP SHOW LOAD.

SHOW LOAD hanya berlaku untuk metode impor asinkron. Tidak berlaku untuk metode sinkron seperti Stream Load.

Outputnya tampak seperti:

*************************** 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}

Membatalkan pekerjaan impor

Batalkan pekerjaan yang berstatus PENDING atau LOADING:

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

Untuk melihat sintaks lengkap, jalankan HELP CANCEL LOAD.

Contoh end-to-end: muat dari OSS

Contoh ini memuat data lineitem TPC-H dari OSS ke StarRocks.

Langkah 1: Buat tabel tujuan.

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"
);

Langkah 2: Kirim pekerjaan impor.

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)
)

Langkah 3: Pantau pekerjaan.

SHOW LOAD WHERE label = 'lineitem'\G

Saat pekerjaan berhasil, output menunjukkan State: FINISHED dan Progress: ETL:100%; LOAD:100%.

Langkah 4: Verifikasi data.

-- Periksa jumlah total baris
SELECT COUNT(*) FROM lineitem;

-- Pratinjau dua baris pertama
SELECT * FROM lineitem LIMIT 2;

Konkurensi

Pekerjaan Broker Load dibagi menjadi tugas, lalu tugas tersebut dibagi lagi menjadi instans yang didistribusikan secara paralel di seluruh backend.

Pemisahan tugas: Satu pekerjaan menghasilkan satu tugas per klausa data_desc yang mengarah ke alamat sumber berbeda atau kumpulan partisi berbeda.

Pemisahan instans: Setiap tugas dibagi menjadi instans berdasarkan tiga parameter frontend:

Total instans = min(ukuran_file / min_bytes_per_broker_scanner, max_broker_concurrency, load_parallel_instance_num × jumlah backend)

Dalam praktiknya, sebagian besar pekerjaan memiliki satu klausa data_desc sehingga hanya menghasilkan satu tugas. Tugas tersebut dibagi menjadi sebanyak mungkin instans sesuai jumlah backend, dengan setiap instans berjalan di backend berbeda.

Pemecahan masalah

Langkah selanjutnya

• Tinjau dokumentasi StarRocks untuk opsi lanjutan LOAD LABEL dan penggunaan HELP BROKER LOAD

• Pantau pekerjaan yang sedang berjalan dengan SHOW LOAD dan tinjau log backend untuk penyetelan performa

• Sesuaikan min_bytes_per_broker_scanner, max_broker_concurrency, dan load_parallel_instance_num di frontend untuk menyetel konkurensi impor sesuai kluster Anda