E-MapReduce:Spark Load

When it's needed

Doris BITMAP columns use roaring bitmaps, which require integer input. When importing data into BITMAP aggregate columns, Spark Load can build a global dictionary to encode raw values as integers during the ETL phase. The dictionary is stored as a Hive table structure mapping original values to encoded values.

How it's built

1. Read source data into a temporary Hive table (hive_table).

2. Deduplicate values and store distinct raw values in distinct_value_table.

3. Create a dict_table with columns for raw values and encoded values.

4. Perform a LEFT JOIN on distinct_value_table and dict_table, encode new values using a window function, and write results back to dict_table.

5. Join dict_table with hive_table to replace raw values with encoded integers.

6. Feed the encoded hive_table into the remaining ETL pipeline for import into Doris.

Hive Bitmap UDF

Spark Load supports directly importing bitmap data generated in Hive into Doris using the Hive Bitmap UDF.

Create a resource

CREATE EXTERNAL RESOURCE resource_name
PROPERTIES
(
  type = spark,
  spark_conf_key = spark_conf_value,
  working_dir = path,
  broker = broker_name,
  broker.property_key = property_value,
  broker.hadoop.security.authentication = kerberos,
  broker.kerberos_principal = doris@YOUR.COM,
  broker.kerberos_keytab = /home/doris/my.keytab
  broker.kerberos_keytab_content = ASDOWHDLAWI********ALDJSDIWALD
)

The following table describes the key parameters:

For a full list of Spark configuration parameters, see Spark Configuration.

Example: YARN cluster mode

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" = "127.0.0.1:9999",
  "spark.hadoop.fs.defaultFS" = "hdfs://127.0.0.1:10000",
  "working_dir" = "hdfs://127.0.0.1:10000/tmp/doris",
  "broker" = "broker0",
  "broker.username" = "user0",
  "broker.password" = "password0"
);

Example: Spark standalone client mode

CREATE EXTERNAL RESOURCE "spark1"
PROPERTIES
(
  "type" = "spark",
  "spark.master" = "spark://127.0.0.1:7777",
  "spark.submit.deployMode" = "client",
  "working_dir" = "hdfs://127.0.0.1:10000/tmp/doris",
  "broker" = "broker1"
);

Example: Kerberos-authenticated cluster

CREATE EXTERNAL RESOURCE "spark_on_kerberos"
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" = "127.0.0.1:9999",
  "spark.hadoop.fs.defaultFS" = "hdfs://127.0.0.1:10000",
  "working_dir" = "hdfs://127.0.0.1:10000/tmp/doris",
  "broker" = "broker0",
  "broker.hadoop.security.authentication" = "kerberos",
  "broker.kerberos_principal" = "doris@YOUR.COM",
  "broker.kerberos_keytab" = "/home/doris/my.keytab"
);

Manage resources

View resources

SHOW RESOURCES
SHOW PROC "/resources"

Ordinary accounts can only see resources for which they have USAGE_PRIV. Root and admin accounts see all resources.

Grant and revoke permissions

-- Grant USAGE_PRIV on a specific resource to a user
GRANT USAGE_PRIV ON RESOURCE "spark0" TO "user0"@"%";

-- Grant USAGE_PRIV on a specific resource to a role
GRANT USAGE_PRIV ON RESOURCE "spark0" TO ROLE "role0";

-- Grant USAGE_PRIV on all resources to a user
GRANT USAGE_PRIV ON RESOURCE * TO "user0"@"%";

-- Grant USAGE_PRIV on all resources to a role
GRANT USAGE_PRIV ON RESOURCE * TO ROLE "role0";

-- Revoke USAGE_PRIV on a specific resource from a user
REVOKE USAGE_PRIV ON RESOURCE "spark0" FROM "user0"@"%";

Drop a resource

DROP RESOURCE resource_name

Syntax

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

-- load_label:
    db_name.label_name

-- data_desc (HDFS file):
    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]

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

-- resource_properties:
    (key2=value2, ...)

Supported data sources are CSV files in HDFS and Hive external tables. Other rules follow Broker Load conventions.

Parameters

Data description parameters

Only CSV files and Hive tables are supported as data sources. Other rules are consistent with Broker Load.

Job parameters

Job parameters (set in PROPERTIES) apply to the entire import job and follow the same rules as Broker Load.

Per-job Spark resource overrides

To override Spark resource settings for a single job without changing the cluster-level resource configuration, specify overrides inline:

WITH RESOURCE 'spark0'
(
  "spark.driver.memory" = "1g",
  "spark.executor.memory" = "3g"
)

Overrides apply only to the current job.

Examples

Example 1: Import from HDFS files

LOAD LABEL db1.label1
(
    DATA INFILE("hdfs://abc.com:8888/user/palo/test/ml/file1")
    INTO TABLE tbl1
    COLUMNS TERMINATED BY ","
    (tmp_c1,tmp_c2)
    SET
    (
        id=tmp_c2,
        name=tmp_c1
    ),
    DATA INFILE("hdfs://abc.com:8888/user/palo/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"
);

Example 2: Import from a Hive table

First, create an external Hive table in Doris:

CREATE EXTERNAL TABLE hive_t1
(
    k1 INT,
    K2 SMALLINT,
    k3 varchar(50),
    uuid varchar(100)
)
ENGINE=hive
properties
(
  "database" = "tmp",
  "table" = "t1",
  "hive.metastore.uris" = "thrift://0.0.0.0:8080"
);

Then submit the load job. The columns in the Doris target table must exist in the external Hive table:

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

Example 3: Import BINARY bitmap data from a Hive table

Use this approach when the source Hive column is of type BINARY and the data was serialized using the org.apache.doris.load.loadv2.dpp.BitmapValue class from spark-dpp.

Create the external Hive table:

CREATE EXTERNAL TABLE hive_t1
(
    k1 INT,
    K2 SMALLINT,
    k3 varchar(50),
    uuid varchar(100) -- BINARY type in the Hive table
)
ENGINE=hive
properties
(
  "database" = "tmp",
  "table" = "t1",
  "hive.metastore.uris" = "thrift://0.0.0.0:8080"
);

Submit the load job using binary_bitmap:

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

Example 4: Import from a partitioned Hive table

Hive table definition:

CREATE TABLE test_partition(
    id int,
    name string,
    age int
)
PARTITIONED BY (dt string)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
STORED AS TEXTFILE;

Doris target table definition:

CREATE TABLE IF NOT EXISTS test_partition_04
(
    dt date,
    id int,
    name string,
    age int
)
UNIQUE KEY(`dt`, `id`)
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
    "replication_allocation" = "tag.location.default: 1"
);

Create the Spark resource and submit the load job:

CREATE EXTERNAL RESOURCE "spark_resource"
PROPERTIES
(
  "type" = "spark",
  "spark.master" = "yarn",
  "spark.submit.deployMode" = "cluster",
  "spark.executor.memory" = "1g",
  "spark.yarn.queue" = "default",
  "spark.hadoop.yarn.resourcemanager.address" = "localhost:50056",
  "spark.hadoop.fs.defaultFS" = "hdfs://localhost:9000",
  "working_dir" = "hdfs://localhost:9000/tmp/doris",
  "broker" = "broker_01"
);

LOAD LABEL demo.test_hive_partition_table_18
(
    DATA INFILE("hdfs://localhost:9000/user/hive/warehouse/demo.db/test/dt=2022-08-01/*")
    INTO TABLE test_partition_04
    COLUMNS TERMINATED BY ","
    FORMAT AS "csv"
    (id,name,age)
    COLUMNS FROM PATH AS (`dt`)
    SET
    (
        dt=dt,
        id=id,
        name=name,
        age=age
    )
)
WITH RESOURCE 'spark_resource'
(
    "spark.executor.memory" = "1g",
    "spark.shuffle.compress" = "true"
)
PROPERTIES
(
    "timeout" = "3600"
);

Global dictionary in load jobs

When the target Doris table has a BITMAP aggregate column, specify bitmap_dict in the SET clause to trigger global dictionary creation:

SET (doris_field_name = bitmap_dict(hive_field_name))