E-MapReduce:Spark Load

ステップ 1：外部 Spark リソースの作成

Spark クラスターを StarRocks の外部リソースとして登録し、フロントエンドノードが ETL ジョブを送信できるようにします。

構文

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

リソースプロパティ

Spark 設定オプションの完全なリストについては、Spark 設定ドキュメントをご参照ください。

例

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

Yarn クラスターモード (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"
);

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

リソースの管理

-- すべてのリソースをリスト表示
SHOW RESOURCES;
SHOW PROC "/resources";

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

通常のアカウントでは、SHOW RESOURCES はアカウントが USAGE-PRIV を持つリソースのみを表示します。root および admin アカウントはすべてのリソースを表示します。

リソースの削除

DROP RESOURCE "resource_name";

ステップ 2：Spark クライアントの設定

フロントエンドノードは spark-submit を使用して Spark Load ジョブを送信するため、Spark クライアントをフロントエンドノードのホストにインストールする必要があります。

Spark 2.4.5 またはそれ以降の 2.x バージョンをダウンロードしてフロントエンドノードに配置します。次に、フロントエンドノードの設定ファイル (fe.conf) で以下のパラメーターを設定します。

1. Spark ホームディレクトリの設定：spark_home_default_dir を、Spark クライアントを配置したディレクトリに設定します。デフォルト値は、フロントエンドノードのルートディレクトリ下の lib/spark2x です。このパラメーターは空白にできません。

2. Spark 依存関係のパッケージ化：Spark クライアントの jars/ フォルダにあるすべての JAR ファイルを ZIP ファイルにパッケージ化します。spark_resource_path をその ZIP ファイルのパスに設定します。空白のままにすると、フロントエンドノードはルートディレクトリ内の lib/spark2x/jars/spark-2x.zip を探します。見つからない場合はエラーが返されます。Spark Load ジョブが送信されると、依存関係パッケージは working_dir/{cluster_id}/ のリモートステージング場所にアップロードされます。ステージング場所は --spark-repository--{resource-name} のフォーマットで命名されます。ディレクトリ構造の例：

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

   デフォルトの依存関係パッケージ名は spark-2x.zip です。フロントエンドノードは、動的パーティションプルーニング (DPP) の依存関係もアップロードします。一度アップロードされると、これらのパッケージは後続のジョブで再利用されます。

ステップ 3：YARN クライアントの設定

フロントエンドノードは YARN コマンドを実行してアプリケーションのステータスを確認したり、アプリケーションを停止したりするため、YARN クライアントもフロントエンドノードにインストールする必要があります。

Hadoop 2.5.2 またはそれ以降の 2.x バージョンをダウンロードし、fe.conf で以下のパラメーターを設定します。

1. YARN 実行可能ファイルのパスの設定：yarn_client_path を YARN バイナリファイルのパスに設定します。デフォルトは、フロントエンドノードのルートディレクトリ下の lib/yarn-client/hadoop/bin/yarn です。

2. (オプション) YARN 設定ディレクトリの設定：フロントエンドノードがアプリケーションのステータスを確認したり、アプリケーションを停止したりすると、デフォルトで lib/yarn-config/ 内に core-site.xml と yarn-site.xml が生成されます。この場所を変更するには、yarn_config_dir を設定します。

インポートジョブの作成

サポートされているデータソース：CSV ファイルと Hive テーブル。

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

ここで、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]

-- Hive テーブルから
DATA FROM TABLE hive_external_tbl
[NEGATIVE]
INTO TABLE tbl_name
[PARTITION (p1, p2)]
[SET (k1=f1(xx), k2=f2(xx))]
[WHERE predicate]

完全な構文については、HELP SPARK LOAD を実行してください。

主要なパラメーター

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

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

例 2：グローバル辞書を使用した Hive テーブルからのインポート

このアプローチは、ターゲットの StarRocks テーブルにビットマップ集約列がある場合に使用します。bitmap_dict() 関数は、グローバル辞書を介して生の Hive 列の値をエンコードされた整数にマッピングします。

1. Hive 外部リソースを作成します。

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

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

   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. インポートジョブを送信します。

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

インポートジョブの確認

SHOW LOAD ORDER BY createtime DESC LIMIT 1\G

出力例：

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

出力フィールド

完全なパラメーターリファレンスについては、「Broker Load」をご参照ください。

完全な構文については、HELP SHOW LOAD を実行してください。

ジョブログの表示

Spark Load のログは、フロントエンドノードのルートディレクトリ下の log/spark_launcher_log/ に保存され、spark-launcher-{load-job-id}-{label}.log のフォーマットで命名されます。ログはデフォルトで 3 日間保持され、関連するインポートメタデータがパージされるときに削除されます。

インポートジョブのキャンセル

FINISHED または CANCELLED 状態でないジョブをキャンセルします。

CANCEL LOAD WHERE LABEL = "label1";

完全な構文については、HELP CANCEL LOAD を実行してください。