E-MapReduce：Stream Load

Istilah

coordinator: Node yang menerima data, mendistribusikannya ke node lain, dan mengembalikan hasil setelah impor selesai.

Prinsip dasar

Stream Load memungkinkan Anda mengirimkan pekerjaan impor melalui permintaan HTTP. Jika permintaan dikirim ke node frontend (FE), FE akan meneruskannya ke node backend (BE) dengan pengalihan HTTP. Anda juga dapat mengirimkan permintaan langsung ke node BE. Node BE bertindak sebagai koordinator untuk membagi data berdasarkan skema tabel dan mendistribusikannya ke node BE terkait. Koordinator kemudian mengembalikan hasil pekerjaan impor kepada Anda.

Properti header yang didukung dapat dilihat pada deskripsi parameter pekerjaan impor dalam tabel berikut. Parameter dikonfigurasi dalam format -H "key1:value1". Jika ada beberapa parameter, gunakan beberapa -H untuk menunjukkan parameter tersebut. Contoh: -H "key1:value1" -H "key2:value2". Dalam mode Stream Load, semua parameter terkait pekerjaan impor dikonfigurasi di header. Tabel berikut menjelaskan parameter tersebut.

Praktik terbaik

Skenario

Stream Load cocok untuk skenario di mana file sumber disimpan di memori atau disk lokal. Sebagai metode impor sinkron, Stream Load memungkinkan Anda mendapatkan hasil pekerjaan impor secara langsung.

Ukuran data

Dalam mode Stream Load, node BE mengimpor dan mendistribusikan data. Disarankan untuk menggunakan Stream Load untuk mengimpor data berukuran antara 1 GB hingga 10 GB. Secara default, ukuran maksimum data yang dapat diimpor adalah 10 GB. Untuk mengimpor file yang lebih besar, modifikasi parameter streaming_load_max_mb node BE. Misalnya, jika ukuran file adalah 15 GB (15.360 MB), atur parameter streaming_load_max_mb menjadi nilai lebih besar dari 15 GB.

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

Periode timeout default untuk pekerjaan impor dalam Stream Load adalah 600 detik. Untuk menyesuaikan periode timeout, ubah nilai parameter timeout node FE di instance StarRocks Anda melalui Konsol EMR.

Contoh lengkap

Data disimpan di direktori /mnt/disk1/customer.tbl klien tertentu. Anda ingin mengimpor data ke tabel customer database stream_load instance StarRocks Anda.

Unduh data: customer.tbl

Jumlah pekerjaan impor dalam mode Stream Load yang dapat diproses secara bersamaan tidak dipengaruhi oleh ukuran instance.

Prosedur:

1. Modifikasi file konfigurasi BE.conf node BE jika ukuran file melebihi batas atas default. Sebagai contoh, atur parameter streaming_load_max_mb menjadi 15360. Unit: MB.

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

2. Di tab Konfigurasi Instance instance Anda, ubah nilai parameter stream_load_default_timeout_second. Dalam contoh ini, atur nilainya menjadi 3600.

3. Buat tabel tujuan bernama 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. Buat pekerjaan impor. Jika kumpulan data berisi banyak data, jalankan operasi di latar belakang.

   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

   Informasi berikut dikembalikan:

   {
       "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
   }

   null

   Jika terjadi kesalahan "ErrorURL": "http://***:8040/api/_load_error_log?file=error_log_***", Anda dapat menjalankan perintah curl untuk melihat detailnya.

Contoh kode untuk integrasi

• Untuk informasi tentang cara menggunakan Java untuk mengirimkan pekerjaan impor dalam Stream Load, lihat stream_load.

• Untuk informasi tentang cara menggunakan Spark untuk mengirimkan pekerjaan impor dalam Stream Load, lihat 01_sparkStreaming2StarRocks.