E-MapReduce:Spark Load

Langkah 1: Buat resource Spark eksternal

Daftarkan kluster Spark sebagai resource eksternal di StarRocks agar node frontend dapat mengirimkan pekerjaan ETL ke kluster tersebut.

Sintaks

CREATE EXTERNAL RESOURCE "resource_name"
PROPERTIES
(
    "type" = "spark",
    "spark.master" = "yarn",
    "spark.submit.deployMode" = "<cluster|client>",
    "spark.hadoop.fs.defaultFS" = "hdfs://<namenode-host>:<port>",
    "spark.hadoop.yarn.resourcemanager.address" = "<resourcemanager-host>:8032",
    "working_dir" = "hdfs://<namenode-host>:<port>/tmp/starrocks",
    "broker" = "<broker-name>",
    "broker.username" = "<username>",
    "broker.password" = "<password>"
);

Properti resource

Untuk daftar lengkap opsi konfigurasi Spark, lihat dokumentasi Konfigurasi Spark.

Contoh

Mode kluster Yarn (standar):

CREATE EXTERNAL RESOURCE "spark0"
PROPERTIES
(
    "type" = "spark",
    "spark.master" = "yarn",
    "spark.submit.deployMode" = "cluster",
    "spark.jars" = "xxx.jar,yyy.jar",
    "spark.files" = "/tmp/aaa,/tmp/bbb",
    "spark.executor.memory" = "1g",
    "spark.yarn.queue" = "queue0",
    "spark.hadoop.yarn.resourcemanager.address" = "resourcemanager_host:8032",
    "spark.hadoop.fs.defaultFS" = "hdfs://namenode_host:9000",
    "working_dir" = "hdfs://namenode_host:9000/tmp/starrocks",
    "broker" = "broker0",
    "broker.username" = "user0",
    "broker.password" = "password0"
);

Mode kluster Yarn dengan Ketersediaan tinggi YARN:

CREATE EXTERNAL RESOURCE "spark1"
PROPERTIES
(
    "type" = "spark",
    "spark.master" = "yarn",
    "spark.submit.deployMode" = "cluster",
    "spark.hadoop.yarn.resourcemanager.ha.enabled" = "true",
    "spark.hadoop.yarn.resourcemanager.ha.rm-ids" = "rm1,rm2",
    "spark.hadoop.yarn.resourcemanager.hostname.rm1" = "host1",
    "spark.hadoop.yarn.resourcemanager.hostname.rm2" = "host2",
    "spark.hadoop.fs.defaultFS" = "hdfs://namenode_host:9000",
    "working_dir" = "hdfs://namenode_host:9000/tmp/starrocks",
    "broker" = "broker1"
);

Mode Ketersediaan tinggi HDFS:

CREATE EXTERNAL RESOURCE "spark2"
PROPERTIES
(
    "type" = "spark",
    "spark.master" = "yarn",
    "spark.hadoop.yarn.resourcemanager.address" = "resourcemanager_host:8032",
    "spark.hadoop.fs.defaultFS" = "hdfs://myha",
    "spark.hadoop.dfs.nameservices" = "myha",
    "spark.hadoop.dfs.ha.namenodes.myha" = "mynamenode1,mynamenode2",
    "spark.hadoop.dfs.namenode.rpc-address.myha.mynamenode1" = "nn1_host:rpc_port",
    "spark.hadoop.dfs.namenode.rpc-address.myha.mynamenode2" = "nn2_host:rpc_port",
    "spark.hadoop.dfs.client.failover.proxy.provider" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider",
    "working_dir" = "hdfs://myha/tmp/starrocks",
    "broker" = "broker2",
    "broker.dfs.nameservices" = "myha",
    "broker.dfs.ha.namenodes.myha" = "mynamenode1,mynamenode2",
    "broker.dfs.namenode.rpc-address.myha.mynamenode1" = "nn1_host:rpc_port",
    "broker.dfs.namenode.rpc-address.myha.mynamenode2" = "nn2_host:rpc_port",
    "broker.dfs.client.failover.proxy.provider" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
);

Mengelola sumber daya

-- Menampilkan semua resource
SHOW RESOURCES;
SHOW PROC "/resources";

-- Memberikan atau mencabut USAGE-PRIV
GRANT USAGE_PRIV ON RESOURCE "spark0" TO "user0"@"%";
GRANT USAGE_PRIV ON RESOURCE "spark0" TO ROLE "role0";
GRANT USAGE_PRIV ON RESOURCE * TO "user0"@"%";
GRANT USAGE_PRIV ON RESOURCE * TO ROLE "role0";
REVOKE USAGE_PRIV ON RESOURCE "spark0" FROM "user0"@"%";
REVOKE USAGE_PRIV ON RESOURCE "spark0" FROM ROLE "role0";

Dengan akun biasa, SHOW RESOURCES hanya menampilkan resource yang memiliki izin USAGE-PRIV untuk akun tersebut. Akun root dan admin melihat semua resource.

Melepaskan sumber daya

DROP RESOURCE "resource_name";

Langkah 2: Konfigurasikan klien Spark

Node frontend menggunakan spark-submit untuk mengirimkan pekerjaan Spark Load, sehingga klien Spark harus diinstal pada host node frontend.

Unduh Spark 2.4.5 atau versi 2.x terbaru dan letakkan di node frontend. Lalu konfigurasikan parameter berikut dalam file konfigurasi node frontend (fe.conf):

1. Tetapkan direktori home Spark Atur spark_home_default_dir ke direktori tempat Anda meletakkan klien Spark. Nilai default adalah lib/spark2x di bawah direktori root node frontend. Parameter ini tidak boleh kosong.

2. Paket dependensi Spark Kemas semua file JAR dari folder jars/ klien Spark ke dalam satu file ZIP. Atur spark_resource_path ke path file ZIP tersebut. Jika dibiarkan kosong, node frontend akan mencari lib/spark2x/jars/spark-2x.zip di direktori root-nya—jika tidak ditemukan, akan muncul error. Saat pekerjaan Spark Load dikirimkan, paket dependensi diunggah ke lokasi staging jarak jauh di working_dir/{cluster_id}/. Lokasi staging dinamai dalam format --spark-repository--{resource-name}. Contoh struktur direktori:

   ---spark-repository--spark0/
      |---archive-1.0.0/
      |   |---lib-990325d2c0d1d5e45bf675e54e44fb16-spark-dpp-1.0.0-jar-with-dependencies.jar
      |   |---lib-7670c29daf535efe3c9b923f778f61fc-spark-2x.zip
      |---archive-1.1.0/
      |   |---lib-64d5696f99c379af2bee28c1c84271d5-spark-dpp-1.1.0-jar-with-dependencies.jar
      |   |---lib-1bbb74bb6b264a270bc7fca3e964160f-spark-2x.zip

   Nama default paket dependensi adalah spark-2x.zip. Node frontend juga mengunggah dependensi Dynamic Partition Pruning (DPP). Setelah diunggah, paket-paket ini digunakan kembali untuk pekerjaan selanjutnya.

Langkah 3: Konfigurasikan klien YARN

Node frontend menjalankan perintah YARN untuk memeriksa status aplikasi dan menghentikan aplikasi, sehingga klien YARN juga harus diinstal pada node frontend.

Unduh Hadoop 2.5.2 atau versi 2.x terbaru dan konfigurasikan parameter berikut dalam fe.conf:

1. Tetapkan path eksekusi YARN Atur yarn_client_path ke path file biner YARN. Default-nya adalah lib/yarn-client/hadoop/bin/yarn di bawah direktori root node frontend.

2. (Opsional) Tetapkan direktori konfigurasi YARN Saat node frontend memeriksa status aplikasi atau menghentikan aplikasi, secara default akan menghasilkan core-site.xml dan yarn-site.xml di lib/yarn-config/. Untuk mengubah lokasi ini, atur yarn_config_dir.

Buat pekerjaan impor

Sumber data yang didukung: file CSV dan tabel Hive.

LOAD LABEL db_name.label_name
    (data_desc, ...)
WITH RESOURCE resource_name
[resource_properties]
[PROPERTIES (key1=value1, ...)]

Dengan data_desc adalah salah satu dari:

-- Dari file HDFS
DATA INFILE ("file_path", ...)
[NEGATIVE]
INTO TABLE tbl_name
[PARTITION (p1, p2)]
[COLUMNS TERMINATED BY separator]
[(col1, ...)]
[COLUMNS FROM PATH AS (col2, ...)]
[SET (k1=f1(xx), k2=f2(xx))]
[WHERE predicate]

-- Dari tabel Hive
DATA FROM TABLE hive_external_tbl
[NEGATIVE]
INTO TABLE tbl_name
[PARTITION (p1, p2)]
[SET (k1=f1(xx), k2=f2(xx))]
[WHERE predicate]

Untuk sintaks lengkap, jalankan HELP SPARK LOAD.

Parameter utama

Contoh 1: Impor dari file CSV HDFS

LOAD LABEL db1.label1
(
    DATA INFILE("hdfs://emr-header-1.cluster-xxx:9000/user/starRocks/test/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/starRocks/test/ml/file2")
    INTO TABLE tbl2
    COLUMNS TERMINATED BY ","
    (col1, col2)
    WHERE col1 > 1
)
WITH RESOURCE 'spark0'
(
    "spark.executor.memory" = "2g",
    "spark.shuffle.compress" = "true"
)
PROPERTIES
(
    "timeout" = "3600"
);

Contoh 2: Impor dari tabel Hive dengan kamus global

Gunakan pendekatan ini ketika tabel StarRocks target memiliki kolom agregasi bitmap. Fungsi bitmap_dict() memetakan nilai kolom Hive mentah ke bilangan bulat terenkripsi melalui kamus global.

1. Buat resource Hive eksternal:

   CREATE EXTERNAL RESOURCE hive0
   PROPERTIES
   (
       "type" = "hive",
       "hive.metastore.uris" = "thrift://emr-header-1.cluster-xxx:9083"
   );

2. Buat tabel Hive eksternal:

   CREATE EXTERNAL TABLE hive_t1
   (
       k1 INT,
       K2 SMALLINT,
       k3 VARCHAR(50),
       uuid VARCHAR(100)
   )
   ENGINE=hive
   PROPERTIES
   (
       "resource" = "hive0",
       "database" = "tmp",
       "table" = "t1"
   );

3. Kirim pekerjaan impor:

   LOAD LABEL db1.label1
   (
       DATA FROM TABLE hive_t1
       INTO TABLE tbl1
       SET
       (
           uuid=bitmap_dict(uuid)
       )
   )
   WITH RESOURCE 'spark0'
   (
       "spark.executor.memory" = "2g",
       "spark.shuffle.compress" = "true"
   )
   PROPERTIES
   (
       "timeout" = "3600"
   );

Lihat pekerjaan impor

SHOW LOAD ORDER BY createtime DESC LIMIT 1\G

Contoh output:

*************************** 1. row ***************************
       JobId: 76391
       Label: label1
       State: FINISHED
    Progress: ETL:100%; LOAD:100%
        Type: SPARK
     EtlInfo: unselected.rows=4; dpp.abnorm.ALL=15; dpp.norm.ALL=28133376
    TaskInfo: cluster:cluster0; 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:49:44
LoadStartTime: 2019-07-27 11:49:44
LoadFinishTime: 2019-07-27 11:50:16
         URL: http://1.1.*.*:8089/proxy/application_1586619723848_0035/
  JobDetails: {"ScannedRows":28133395,"TaskNumber":1,"FileNumber":1,"FileSize":200000}

Bidang output

Untuk referensi parameter lengkap, lihat Broker Load.

Untuk sintaks lengkap, jalankan HELP SHOW LOAD.

Lihat log pekerjaan

Log Spark Load disimpan di log/spark_launcher_log/ di bawah direktori root node frontend, dengan nama dalam format spark-launcher-{load-job-id}-{label}.log. Log disimpan selama tiga hari secara default dan dihapus saat metadata impor terkait dipurge.

Batalkan pekerjaan impor

Batalkan pekerjaan yang tidak berada dalam status FINISHED atau CANCELLED:

CANCEL LOAD WHERE LABEL = "label1";

Untuk sintaks lengkap, jalankan HELP CANCEL LOAD.