欢迎您访问365答案网,请分享给你的朋友!
生活常识 学习资料

第3.8章:StarRocks数据导入--SparkLoad

时间:2023-04-18

Spark Load是通过外部的Spark资源实现对导入数据的预处理,进而提高StarRocks大数据量的导入性能,同时也可以节省StarRocks集群的计算资源。Spark Load的操作本身不复杂,但涉及的技术栈比较多,架构相对较重,所以主要用于初次迁移、大数据量导入等场景(数据量可到TB级别)。

Spark Load的特点在于其引入了外部Spark集群,让我们可以方便的使用Spark 执行 ETL 完成对导入数据的预处理,包括全局字典构建(BITMAP类型)、分区、排序、聚合等。

StarRocks官网文档对Spark Load的介绍地址如下:

Spark Load @ SparkLoad @ StarRocks DocsSpark Loadhttps://docs.starrocks.com/zh-cn/main/loading/SparkLoad通常来说,Spark Load的整体逻辑是:读Hadoop数据源中的数据文件-->使用Spark自己的计算资源进行ETL-->将ETL后的文件保存到计算资源中配置的Hadoop中-->自动使用Broker Load导入上一步的中间文件到StarRocks。

一、集群准备

演示需要用到的服务器及其上部署的主要服务见下表:

节点IP

部署服务

端口

版本

说明

192.168.110.11

[node01]

NameNode

9000

hadoop-3.3.0

core-site.xml中fs.defaultFS参数的端口

DataNode

Hadoop所用端口均为默认

NodeManager

192.168.110.12

[node02]

DataNode

hadoop-3.3.0

ResourceManager

8032

RM的applications manager(ASM)端口

NodeManager

Spark

spark-2.4.8-bin-hadoop2.7

作为计算资源

Scala

scala-2.11.12

192.168.110.13

[node03]

SecondaryNameNode

hadoop-3.3.0

DataNode

NodeManager

192.168.110.98

[starrocks]

FE

9030

StarRocks 1.19.0

StarRocks所用端口均为默认端口

BE

Broker

Broker名称:hdfs_broker

mysql-client

mysql-client-5.7.35

Spark客户端

每个FE节点都需要配置

spark-2.4.8-bin-hadoop2.7

无需启动,可简单理解为是使用Spark Load时FE的依赖包

Yarn客户端

hadoop-2.10.0

本次演示,我们模拟从任意Hadoop中读取数据,Spark对数据预处理后会将中间文件保存在配置的Hadoop中,然后自动通过Broker Load的方式完成导入。整个操作其实涉及到个两套Hadoop集群,一套是用来做数据源的,另一套是用来存储Spark预处理的数据的,这两套Hadoop集群可以相同也可以不相同,咱们上面使用的是相同的,其版本没有什么要求。对于上表中的服务配置还有一些需要说明的地方:

1、为方便演示,我们在starrocks节点单机部署了StarRocks;

2、starrocks节点的Spark客户端需要使用Spark 2.4.5或以上的Spark2官方版本(不支持Spark3);

3、node02节点用来作为计算资源的Spark版本建议与starrocks节点的“Spark客户端”版本保持一致,同时需要以spark on yarn的模式运行;

4、starrocks节点的Yarn客户端需要使用Hadoop 2.5.2或以上的hadoop官方版本,支持Hadoop 3,但更推荐使用Hadoop 2版本,因为测试更充分;

5、node01~03节点的Hadoop集群对Hadoop版本没有特别要求,但都需要配置环境变量HADOOP_HOME,例如:

vim /etc/profile

##HADOOP_HOME

export HADOOP_HOME=/opt/module/hadoop-3.3.0

export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin

6、以上所有节点均部署了Oracle jdk 1.8.0_201(推荐使用Oracle jdk 1.8+),并配置了环境变量;

7、StarRocks集群中每个FE节点都需要配置Spark客户端与Yarn客户端,多FE时注意不要漏配(因为无法确定请求从哪个FE上发出,所以需要对每个FE都进行配置);

8、StarRocks集群中使用Spark客户端时不需要部署Scala,也不需要配置Spark、Yarn或Hadoop的环境变量。

在starrocks节点(192.168.110.98)上,各应用所在的路径为:

应用

部署路径

FE

/opt/module/starrocks/fe

BE

/opt/module/starrocks/be

Broker

/opt/module/starrocks/apache_hdfs_broker

Spark客户端

/opt/module/spark-2.4.8-bin-hadoop2.7

Yarn客户端

/opt/module/hadoop-2.10.0

二、导入介绍

Spark Load的整体操作可以分为五个步骤:

配置外部Spark资源-->FE配置Spark客户端-->FE配置YARN客户端-->配置计算资源Spark集群-->执行导入任务,我们逐项展开。

2.1 配置外部Spark资源

在starrocks节点,使用mysql-client访问StarRocks服务,为方便演示,我们使用root用户登录(密码也为root):

[root@starrocks ~]# mysql -h192.168.110.98 -P9030 -uroot -proot

要使用Spark Load,我们首先需要在StarRocks集群中创建Spark计算资源。也就是说,这里面配置的,就是我们需要使用的外部Spark计算集群。根据当前测试环境的配置情况,我们先创建名为spark_resource0的资源:

mysql> CREATE EXTERNAL RESOURCE "spark_resource0"

PROPERTIES

(

"type" = "spark",

"spark.master" = "yarn",

"spark.submit.deployMode" = "cluster",

"spark.executor.memory" = "1g",

"spark.hadoop.yarn.resourcemanager.address" = "192.168.110.12:8032",

"spark.hadoop.fs.defaultFS" = "hdfs://192.168.110.11:9000",

"working_dir" = "hdfs://192.168.110.11:9000/tmp/starrocks/sparketl",

"broker" = "hdfs_broker"

);

创建语句中主要参数的释义如下:

type:资源类型,必填,目前仅支持spark;

spark.master:必填,目前支持yarn;

spark.submit.deployMode:Spark程序的部署模式,必填,支持cluster,client两种;

spark.hadoop.yarn.resourcemanager.address:单点resource manager地址,端口需使用ASM端口,默认为8032;

spark.hadoop.fs.defaultFS:master为yarn时必填;

working_dir:Spark作为ETL资源时使用的目录,必填;

broker:StarRocks集群中broker的名字,在StarRocks中可以通过"show broker; "命令查看,Spark作为ETL资源使用时必填。

PROPERTIES部分的参数实际是Spark的参数,可以参考Spark文档中的参数介绍:https://spark.apache.org/docs/latest/configuration.htmlhttps://spark.apache.org/docs/latest/configuration.html

创建完成后,我们可以查看StarRocks中已经创建的资源:

mysql> SHOW RESOURCES;

对于不需要或者创建错误的资源,我们也可以执行删除,例如:

mysql> DROp RESOURCE spark_resource1;

创建好的外部资源,我们还需要授权给后面执行Spark Load命令的用户或角色,比如我们将资源spark_resource0的USAGE_PRIV权限授权给root用户(root用户本身便拥有所有资源使用权限,这里仅演示语法):

mysql> GRANT USAGE_PRIV ON RESOURCE "spark_resource0" TO "root"@"%";

2.2 FE配置Spark客户端

在刚接触Spark Load时,社区不时有同学疑惑,为FE配置Spark和Yarn客户端?那是不是说StarRocks集群要和Hadoop集群以及Spark集群耦合在一起?要混合部署在相同机器上吗?

其实不是的,这里的“客户端”可以理解为“依赖文件”,我们只需要在FE节点放置Spark和Yarn的二进制文件并简单配置就可以,并不需要启动它们,完全可以认为它们是StarRocks执行Spark Load时FE所需要的依赖。

在执行Spark Load时,FE底层是通过执行spark-submit的命令去提交spark任务,所以是需要为FE配置spark客户端的。同样,FE底层通过执行yarn命令去获取正在运行的application的状态,以及杀死application,因此也需要为FE配置yarn客户端。

当前测试环境只有一个FE,我们就将从官方下载好的spark-2.4.8-bin-hadoop2.7.tgz(https://archive.apache.org/dist/spark/spark-2.4.8/)解压至预定的目录,例如当前环境的路径为:/opt/module/spark-2.4.8-bin-hadoop2.7。

然后,我们还需要将spark中jars文件夹内所有的jar包打包为zip文件并命名为spark-2x.zip,例如当前环境下,我们执行打包命令:

[root@starrocks ~]# cd /opt/module/spark-2.4.8-bin-hadoop2.7

[root@starrocks spark-2.4.8-bin-hadoop2.7]# zip -q -r jars/spark-2x.zip jars/*.jar

[root@starrocks spark-2.4.8-bin-hadoop2.7]# ls ./jars/ | grep spark-2x.zip

spark-2x.zip

[root@starrocks spark-2.4.8-bin-hadoop2.7]# cd jars

[root@starrocks jars]# pwd

/opt/module/spark-2.4.8-bin-hadoop2.7/jars

前面的操作我们得到Spark客户端的路径、Spark依赖包及其路径,接下来,我们就需要在FE的配置文件中添加这里的配置:

[root@starrocks ~]# vim /opt/module/starrocks/fe/conf/fe.conf

加入配置:

#配置SPARK-HOME环境变量

spark_home_default_dir = /opt/module/spark-2.4.8-bin-hadoop2.7

#配置SPARK依赖包

spark_resource_path = /opt/module/spark-2.4.8-bin-hadoop2.7/jars/spark-2x.zip

2.3 FE配置YARN客户端

在配置完Spark客户端后,我们继续为FE配置Yarn客户端。首先还是将下载好的hadoop-2.10.0.tar.gz(https://archive.apache.org/dist/hadoop/common/hadoop-2.10.0/)解压至预置路径,比如当前环境的:/opt/module/hadoop-2.10.0。

接下来,我们还需要修改一下yarn的配置文件:

[root@starrocks ~]# vim /opt/module/starrocks/fe/lib/yarn-client/hadoop/libexec/hadoop-config.sh

在其首行添加:

#配置JAVA_HOME

export JAVA_HOME=/usr/java/jdk1.8.0_201

添加完成后,我们再次编辑fe.conf:

[root@starrocks ~]# vim /opt/module/starrocks/fe/conf/fe.conf

加入配置:

#配置YARN可执行文件路径

yarn_client_path = /opt/module/hadoop-2.10.0/bin/yarn

添加完成后退出。

修改FE的配置文件后,我们需要重启FE让配置生效。

-----------------------------------------------------------------------------------------------------------------------

客户端配置补充说明

对于前面的步骤二和步骤三的配置,我们也可以采用另一套方案,即不修改FE配置文件,而是通过创建软链接的方式,将Spark目录和Yarn目录直接映射至FE配置参数的默认路径下。

配置项spark_home_default_dir默认为FE根目录下的 lib/spark2x路径,spark_resource_path参数默认为FE根目录下的lib/spark2x/jars/spark-2x.zip。

那么我们就可以使用软连接配置Spark客户端:

[root@starrocks ~]# cd /opt/module/starrocks/fe/lib

[root@starrocks ~]# ln -s /opt/module/spark-2.4.8-bin-hadoop2.7 spark2x

yarn_client_path默认为FE根目录下的lib/yarn-client/hadoop/bin/yarn路径,同理,我们也可以这样配置YARN客户端:

[root@starrocks ~]# mkdir /opt/module/starrocks/fe/lib/yarn-client

[root@starrocks ~]# cd /opt/module/starrocks/fe/lib/yarn-client

[root@starrocks ~]# ln -s /opt/module/hadoop-2.10.0 hadoop

这样配置完成后,我们就无需重启集群的FE服务了。

-----------------------------------------------------------------------------------------------------------------------

2.4 配置计算资源Spark集群

在node02节点,我们进行了Spark on Yarn单机环境搭建,简单的说明一下部署过程:

1、解压Scala文件并配置环境变量

[root@node02 ~]# vim /etc/profile

添加:

export SCALA_HOME=/opt/module/scala-2.11.12

export PATH=$PATH:$SCALA_HOME/bin

2、分发Spark文件并在profile中配置环境变量

export SPARK_HOME=/opt/module/spark-2.4.8-bin-hadoop2.7

export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin

完成后执行source命令使环境变量生效:

[root@node02 ~]# source /etc/profile

3、修改Spark配置文件

[root@node02 ~]# cd /opt/module/spark-2.4.8-bin-hadoop2.7/conf

[root@node02 ~]# cp spark-env.sh.template spark-env.sh

[root@node02 ~]# vim spark-env.sh

在末尾添加下列配置:

# Hadoop配置文件目录

export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop

# YARN配置文件目录

export YARN_CONF_DIR=$HADOOP_HOME/etc/hadoop

# SPARK目录

export SPARK_HOME=/opt/module/spark-2.4.8-bin-hadoop2.7

# SPARK执行文件目录

export PATH=$SPARK_HOME/bin:$PATH

# 使Spark可读写HDFS中的数据

export SPARK_DIST_CLASSPATH=$(hadoop classpath)

# 配置JAVA_HOME

export JAVA_HOME=/usr/java/jdk1.8.0_201

4、验证集群状态

我们首先确认hdfs和yarn都已启动,然后运行自带的例子SparkPi:

[root@node02 ~]# cd $SPARK_HOME

[root@node02 ~]#./bin/spark-submit --class org.apache.spark.examples.SparkPi --master yarn --deploy-mode cluster --driver-memory 1G --num-executors 3 --executor-memory 1G --executor-cores 1  ./examples/jars/spark-examples_2.11-2.4.8.jar 100

如果看到控制台最后出现“final status: SUCCEEDED”,说明运行成功。

5、启动Spark服务

[root@node02 ~]# cd $SPARK_HOME

[root@node02 ~]#./sbin/start-all.sh

停止服务的命令为:./sbin/stop-all.sh

2.5 执行导入任务

到此,配置部分的准备工作已经完成,接下来继续演示一个简单的例子。我们使用Spark Load将HDFS中customer.csv文件的数据导入至StarRocks ods_data_load库的customer表,假设文件customer.csv的路径为:

hdfs://192.168.110.11:9000/loong/customer.csv(数据文件见评论区度盘),其内共有3W行数据,数据格式如下:

1|Customer#000000001|j5JsirBM9P|MOROCCO  0|MOROCCO|AFRICA|25-989-741-2988|BUILDING

2|Customer#000000002|487LW1dovn6Q4dMVym|JORDAN   1|JORDAN|MIDDLE EAST|23-768-687-3665|AUTOMOBILE

3|Customer#000000003|fkRGN8n|ARGENTINA7|ARGENTINA|AMERICA|11-719-748-3364|AUTOMOBILE

………………

我们先在StarRocks中创建对应数据的数据库及数据表,例如:

mysql> create database if not exists ods_data_load;

mysql> CREATE TABLE IF NOT EXISTS ods_data_load.`customer` (

  `c_custkey` int(11) NOT NULL COMMENT "",

  `c_name` varchar(26) NOT NULL COMMENT "",

  `c_address` varchar(41) NOT NULL COMMENT "",

  `c_city` varchar(11) NOT NULL COMMENT "",

  `c_nation` varchar(16) NOT NULL COMMENT "",

  `c_region` varchar(13) NOT NULL COMMENT "",

  `c_phone` varchar(16) NOT NULL COMMENT "",

  `c_mktsegment` varchar(11) NOT NULL COMMENT ""

) ENGINE=OLAP

DUPLICATE KEY(`c_custkey`)

COMMENT "OLAP"

DISTRIBUTED BY HASH(`c_custkey`) BUCKETS 12

PROPERTIES (

"replication_num" = "1"

);

准备工作全部完成后,启动Spark Load开始导入数据:

mysql> LOAD LABEL ods_data_load.label111301

(

    DATA INFILE("hdfs://192.168.110.11:9000/loong/customer.csv")

    INTO TABLE customer

    COLUMNS TERMINATED BY "|"

    (c_custkey,c_name,c_address,c_city,c_nation,c_region,c_phone,c_mktsegment)

)

WITH RESOURCE 'spark_resource0'

(

    "spark.executor.memory" = "2g",

    "spark.shuffle.compress" = "true"

)

PROPERTIES

(

    "timeout" = "3600"

);

LOAD语句的写法与Broker Load的写法非常相似,我们可以通过set或where语句来进行ETL,这部分可以参考上一篇Broker Load的介绍,这里不再赘述。和Broker Load最大的不同是,Spark Load的ETL是使用我们配置的Spark资源进行的,Broker Load等其他导入方式是使用StarRocks自身的计算资源,这也是Spark Load的设计核心所在。

WITH RESOURCE中的参数是Spark的资源参数,当我们有临时性的需求比如增加任务使用的内存资源时,我们就不需要修改Spark configs,可以在这里直接设置,该设置仅对本次任务生效,并不影响StarRocks集群中现有的配置。

执行任务后,我们同样可以使用show命令查看导入任务:

mysql> use ods_data_load;

mysql> show load order by createtime desc limit 1G

*************************** 1、row ***************************

         JobId: 13046

         Label: label111301

         State: FINISHED

      Progress: ETL:100%; LOAD:100%

          Type: SPARK

       EtlInfo: unselected.rows=0; dpp.abnorm.ALL=0; dpp.norm.ALL=30000

      TaskInfo: cluster:spark_resource0; timeout(s):3600; max_filter_ratio:0.0

      ErrorMsg: NULL

    CreateTime: 2021-11-13 12:34:34

  EtlStartTime: 2021-11-13 12:34:46

 EtlFinishTime: 2021-11-13 12:35:01

 LoadStartTime: 2021-11-13 12:35:01

LoadFinishTime: 2021-11-13 12:35:04

         URL: http://node02:8088/proxy/application_1636629462368_0009/

    JobDetails: {"Unfinished backends":{"00000000-0000-0000-0000-000000000000":[]},"ScannedRows":30000,"TaskNumber":1,"All backends":{"00000000-0000-0000-0000-000000000000":[-1]},"FileNumber":1,"FileSize":2807046}

任务状态中我们需要关注的参数主要有:

State:导入任务当前所处的阶段。任务提交之后状态为PENDING,提交Spark ETL之后状态变为ETL,ETL完成之后FE调度BE执行push操作状态变为LOADING,push完成并且版本生效后状态变为FINISHED。导入任务的最终阶段有两个:CANCELLED和FINISHED,当Load job处于这两个阶段时导入完成。其中CANCELLED为导入失败,FINISHED为导入成功。

Progress:导入任务的进度描述。分为两种进度:ETL和LOAD,对应了导入流程的两个阶段ETL和LOADING。LOAD的进度范围为:0~100%。

LOAD进度 = 当前已完成所有replica导入的tablet个数 / 本次导入任务的总tablet个数* 100%。

如果所有导入表均完成导入,此时LOAD的进度为99%,导入进入到最后生效阶段,整个导入完成后,LOAD的进度才会改为100%。导入进度并不是线性的。所以如果一段时间内进度没有变化,并不代表导入没有在执行。

TaskInfo:显示作业的参数配置,比如使用的资源、超时时间和容错率等等。

JobDetails:显示作业的详细运行状态,包括导入文件的个数、总大小(字节)、子任务个数、已处理的原始行数等。

URL:可复制输入到浏览器,跳转至相应application的web界面。

当State状态为FINISHED时,证明导入任务已经完成,我们可以在StarRocks中验证数据的导入情况:

mysql> use ods_data_load;

……

mysql> select count(1) from customer;

+--------------+

|     count(1) |

+--------------+

|      30000   |

+--------------+

若任务长时间处于PENDING状态,说明任务可能是有问题的,我们可以取消该导入任务:

CANCEL LOAD FROM ods_data_load WHERe LABEL = "label10086";

三、问题排查

排查问题的方法主要就是看报错和日志,Spark Load任务的日志可以从三个途径查看:

1、FE根目录fe/log/spark_launcher_log/下存放spark客户端的提交日志;

2、log/fe.log中可能存在Spark状态的日志;

3、Hadoop集群的日志文件夹下的userlogs目录存放有作业具体失败原因。

Copyright © 2016-2020 www.365daan.com All Rights Reserved. 365答案网 版权所有 备案号:

部分内容来自互联网,版权归原作者所有,如有冒犯请联系我们,我们将在三个工作时内妥善处理。