E-MapReduce:Spark Load

必要なタイミング

Doris の BITMAP 列は Roaring Bitmap を使用しており、整数入力が必要です。BITMAP 集計列へのデータインポート時、Spark Load は ETL フェーズ中に生の値を整数にエンコードするためのグローバル辞書を構築できます。この辞書は、元の値とエンコード済み値をマッピングする Hive テーブル構造として保存されます。

構築方法

1. ソースデータを一時的な Hive テーブル（hive_table）に読み込みます。

2. 値を重複排除し、一意な生の値を distinct_value_table に格納します。

3. 生の値とエンコード済み値の両方を含む列を持つ dict_table を作成します。

4. distinct_value_table と dict_table の間で LEFT JOIN を実行し、ウィンドウ関数を使用して新しい値をエンコードし、結果を dict_table に書き戻します。

5. dict_table と hive_table を結合し、生の値をエンコード済み整数に置き換えます。

6. エンコード済みの hive_table を残りの ETL パイプラインに入力し、Doris にインポートします。

Hive Bitmap UDF

Spark Load は、Hive Bitmap UDF を使用して Hive で生成されたビットマップデータを Doris に直接インポートすることをサポートしています。

リソースの作成

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
)

以下の表は主なパラメーターを説明しています。

Spark 構成パラメーターの完全なリストについては、「Spark Configuration」をご参照ください。

例：YARN クラスターモード

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

例：Spark スタンドアロンクライアントモード

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

例：Kerberos 認証付きクラスター

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

リソースの管理

リソースの表示

SHOW RESOURCES
SHOW PROC "/resources"

通常のアカウントは、USAGE_PRIV 権限を持つリソースのみを表示できます。root および admin アカウントはすべてのリソースを表示できます。

権限の付与および取り消し

-- 特定のリソースに対する USAGE_PRIV をユーザーに付与
GRANT USAGE_PRIV ON RESOURCE "spark0" TO "user0"@"%";

-- 特定のリソースに対する USAGE_PRIV をロールに付与
GRANT USAGE_PRIV ON RESOURCE "spark0" TO ROLE "role0";

-- すべてのリソースに対する USAGE_PRIV をユーザーに付与
GRANT USAGE_PRIV ON RESOURCE * TO "user0"@"%";

-- すべてのリソースに対する USAGE_PRIV をロールに付与
GRANT USAGE_PRIV ON RESOURCE * TO ROLE "role0";

-- 特定のリソースに対する USAGE_PRIV をユーザーから取り消し
REVOKE USAGE_PRIV ON RESOURCE "spark0" FROM "user0"@"%";

リソースの削除

DROP RESOURCE resource_name

構文

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

-- load_label:
    db_name.label_name

-- data_desc (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]

-- data_desc (Hive テーブル):
    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, ...)

サポートされるデータソースは、HDFS 内の CSV ファイルおよび Hive 外部テーブルです。その他の規則は Broker Load の規則に従います。

パラメーター

データ記述パラメーター

データソースとして CSV ファイルおよび Hive テーブルのみがサポートされています。その他の規則は Broker Load と同一です。

ジョブパラメーター

ジョブパラメーター（PROPERTIES 内で設定）はインポートジョブ全体に適用され、Broker Load と同様の規則に従います。

ジョブ単位の Spark リソース設定のオーバーライド

クラスターレベルのリソース構成を変更せずに、単一のジョブに対して Spark リソース設定をオーバーライドするには、インラインで指定します。

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

オーバーライドは現在のジョブにのみ適用されます。

例

例 1：HDFS ファイルからのインポート

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

例 2：Hive テーブルからのインポート

まず、Doris 内に外部 Hive テーブルを作成します。

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

次に、ロードジョブを送信します。Doris のターゲットテーブルの列は、外部 Hive テーブル内に存在している必要があります。

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

例 3：Hive テーブルからの BINARY ビットマップデータのインポート

ソース Hive 列が BINARY 型で、spark-dpp の org.apache.doris.load.loadv2.dpp.BitmapValue クラスを使用してシリアル化されたデータの場合にこの方法を使用します。

外部 Hive テーブルを作成します。

CREATE EXTERNAL TABLE hive_t1
(
    k1 INT,
    K2 SMALLINT,
    k3 varchar(50),
    uuid varchar(100) -- Hive テーブルでは BINARY 型
)
ENGINE=hive
properties
(
  "database" = "tmp",
  "table" = "t1",
  "hive.metastore.uris" = "thrift://0.0.0.0:8080"
);

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

例 4：パーティション化された Hive テーブルからのインポート

Hive テーブル定義：

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

Doris ターゲットテーブル定義：

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

Spark リソースを作成し、ロードジョブを送信します。

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

ロードジョブにおけるグローバル辞書

ターゲット Doris テーブルに BITMAP 集計列がある場合、SET 句で bitmap_dict を指定すると、グローバル辞書の作成がトリガーされます。

SET (doris_field_name = bitmap_dict(hive_field_name))