华为云用户手册

配置场景 在Spark作业的运行过程中，当作业失败时，系统可能无法及时、完整地清理作业所产生的文件，从而造成文件残留。随着时间的推移，这些残留文件会不断累积，持续占用磁盘空间。一旦磁盘空间被大量占用，就极易触发磁盘空间告警，严重时甚至会影响到整个系统的正常运行。 为了有效规避此类问题，有必要对这些残留文件进行定时清理。通过设置合理的定时清理任务，能够及时清除那些无用的残留文件，确保磁盘空间得到合理利用，保障系统的稳定与高效运行。

约束与限制 本章节仅适用于 MRS 3.3.1-LTS及之后版本。 本特性需要启动Spark JD BCS erver服务，借助JDB CS erver服务常驻进程的能力定期清理残留文件。 本特性需要同时配置修改Spark客户端参数、 Spark JDBCServer服务端参数。 支持清理目录： /user/用户/.sparkStaging/ /tmp/hive-scratch/用户 本特性仅支持使用Yarn作为资源调度器的场景。

使用约束 本章节仅适用于MRS 3.5.0及之后版本。 不支持Hudi表的脱敏。 不支持涉及直接读写HDFS的操作的脱敏。 不支持复杂数据类型array、map、struct的脱敏。 自定义脱敏策略仅支持字符串类型字段，且脱敏后的值会强制设置为“***”。 脱敏策略传递时，若目标表已有脱敏策略且与源表有冲突，则目标表脱敏策略强制重置为“Custom:“***””。 简单的未提交Yarn任务的查询操作，脱敏结果与Ranger上配置的脱敏策略一致，Customer类型脱敏策略输出结果为“***”。简单的查询操作例如：select * from 表名;、select * from 表名 limit xxx;。 提交Yarn任务的复杂查询，字符串字段类型脱敏结果与Ranger上配置的脱敏策略一致，其他类型字段脱敏都按“Nullify”脱敏策略处理。

操作步骤 以Hive客户端安装用户登录源端集群安装客户端的节点。 执行以下命令，切换到客户端安装目录，例如安装目录为“/opt/client”，请用户根据实际情况修改。 cd /opt/client 执行以下命令配置环境变量。 source bigdata_env 如果集群开启了Kerberos认证，执行以下命令认证用户，否则跳过此步骤。 kinit Hive业务用户 执行以下命令登录源端集群的Hive客户端。 beeline 执行以下命令创建表“export_test”。 create table export_test(id int); 执行以下命令向表“export_test”中插入数据。 insert into export_test values(123); 在目标集群重复执行1-4，并执行以下命令创建存放表“export_test”导出后的HDFS路径。 hdfs dfs -mkdir /tmp/export 执行以下命令登录目标集群的Hive客户端。 beeline 导入导出表“export_test”。 使用Hive Import/Export对表数据迁移时，支持以下几种场景，可以根据实际情况选择合适的导入导出方式。 场景一：简单导出导入 在源端集群执行以下命令将表“export_test”的元数据和业务数据导出到8创建的目录下。 export table export_test to 'hdfs://haclusterX/tmp/export'; 在目标集群执行以下命令将10.a导出的表数据导入到表“export_test”中。 import from '/tmp/export'; 场景二 ：在导入时重命名表 在源端集群执行以下命令将表“export_test”的元数据和业务数据导出到8创建的目录下。 export table export_test to 'hdfs://haclusterX/tmp/export'; 在目标集群执行以下命令将10.a导出的表数据导入到表“import_test”中。 import table import_test from '/tmp/export'; 场景三：导出分区数据并导入 在源端集群执行以下命令将表“export_test”的pt1和pt2分区导出到8创建的目录下。 export table export_test partition (pt1="in", pt2="ka") to 'hdfs://haclusterX/tmp/export'; 在目标集群执行以下命令将10.a导出的表数据导入到表“export_test”中。 import from '/tmp/export'; 场景四：导出表数据并且将该数据导入到分区中 在源端集群执行以下命令将表“export_test”的元数据和业务数据导出到8创建的目录下。 export table export_test to 'hdfs://haclusterX/tmp/export'; 在目标集群执行以下命令将10.a导出的表数据导入到表“import_test”的pt1和pt2分区中。 import table import_test partition (pt1="us", pt2="tn") from '/tmp/export'; 场景五：导入表数据时指定表的Location 在源端集群执行以下命令将表“export_test”的元数据和业务数据导出到8创建的目录下。 export table export_test to 'hdfs://haclusterX/tmp/export'; 在目标集群执行以下命令将10.a导出的表数据导入到表“import_test”中，且该表的Location为“/tmp/export”。 import table import_test from '/tmp' location '/tmp/export'; 场景六：导入表数据为外部表 在源端集群执行以下命令将表“export_test”的元数据和业务数据导出到8创建的目录下。 export table export_test to 'hdfs://haclusterX/tmp/export'; 在目标集群执行以下命令将10.a导出的表数据导入到外部表“import_test”中。 import external table import_test from '/tmp/export'; “haclusterX”为新增的自定义参数“dfs.namenode.rpc-address.haclusterX”中的“haclusterX”。 导入数据时需注意： 使用import from '/tmp/export';命令导入表是没有指定表名的场景，该场景导入的数据会保存到与源表名相同的表路径下，需注意以下两点： 如果目标集群上不存在与源集群上同名的表，在导入表的过程中会创建该表。 如果目标集群上已存在与源集群上同名的表，该表对应的HDFS目录下必须为空，否则导入失败。 使用import external table import_test from '/tmp/export';命令导入表会将导出的表导入到指定的表中，需注意以下两点： 如果目标集群上不存在与指定的表名相同的表，在导入表的过程中会创建该表。 如果目标集群上已存在与指定的表名相同的表，该表对应的HDFS目录下必须为空，否则导入失败。

操作步骤 以客户端安装用户，登录安装客户端的节点，客户端安装详细操作请参见安装客户端（3.x及之后版本）。 执行以下命令切换到客户端目录，客户端安装目录如：/opt/client。 cd /opt/client 执行以下命令配置环境变量。 source bigdata_env 如果当前集群已启用Kerberos认证，执行以下命令认证当前用户，当前用户需要具有创建Hive表的权限，具体请参见创建角色配置拥有对应权限的角色，参考创建用户为用户绑定对应角色。如果当前集群未启用Kerberos认证，则无需执行此命令。 kinit MRS集群用户 例如，kinit hiveuser。 运行Hive客户端命令，实现A业务。 内部表操作 执行以下命令登录Hive客户端命令行： beeline 根据表1创建用户信息表user_info并添加相关数据，例如： 创建表user_info： create table user_info(id string,name string,gender string,age int,addr string); 向表user_info中插入数据： insert into table user_info(id,name,gender,age,addr) values("12005000201","A","男",19,"A城市"); 在用户信息表user_info中新增用户的学历、职称信息。 以增加编号为12005000201的用户的学历、职称信息为例，其他用户类似。 alter table user_info add columns(education string,technical string); 根据用户编号查询用户姓名和地址。 以查询编号为12005000201的用户姓名和地址为例，其他用户类似。 select name,addr from user_info where id='12005000201'; 删除用户信息表。 drop table user_info; 执行以下命令退出Hive客户端。 !q 外部分区表操作 可使用insert语句直接向外部表中插入数据，也可以使用load data命令导入HDFS中的文件数据到外部表中。如果需要使用load data命令导入文件数据，需执行以下操作： 根据表1数据创建文件。例如，文件名为“txt.log”，以空格拆分字段，以换行符作为行分隔符。 执行以下命令上传文件至HDFS中，例如“/tmp”目录下。 hdfs dfs -put txt.log /tmp 执行以下命令创建外部表数据存储路径： 创建“/hive”目录： hdfs dfs -mkdir /hive/ 创建“/hive/user_info”目录： hdfs dfs -mkdir /hive/user_info 执行以下命令登录Hive客户端命令行： beeline 执行以下命令创建表： create external table user_info(id string,name string,gender string,age int,addr string) partitioned by(year string) row format delimited fields terminated by ' ' lines terminated by '\n' stored as textfile location '/hive/user_info'; 其中： fields terminated：表示分隔的字符，如按空格分隔，则值为' '。 lines terminated：表示分行的字符，如按换行分隔，则值为'\n'。 /hive/user_info：存储表user_info数据的HDFS路径。 导入数据。 使用insert语句插入数据，以插入编号为12005000201的用户相关信息为例，其他用户类似。 insert into user_info partition(year="2018") values ("12005000201","A","男",19,"A城市"); 使用load data命令导入文件数据。 load data inpath '/tmp/txt.log' into table user_info partition (year='2011'); 其中，“/tmp/txt.log”为5.a上传至HDFS的数据文件。 执行以下命令查询导入数据。 select * from user_info; 执行以下命令删除用户信息表。 drop table user_info; 执行以下命令退出Hive客户端。 !q

背景信息 假定用户开发一个应用程序，用于管理企业中的使用A业务的用户信息，使用Hive客户端实现A业务操作流程如下： 普通表的操作： 创建用户信息表user_info。 在用户信息中新增用户的学历、职称信息。 根据用户编号查询用户姓名和地址。 A业务结束后，删除用户信息表。 表1 用户信息 编号 姓名 性别 年龄 地址 12005000201 A 男 19 A城市 12005000202 B 女 23 B城市 12005000203 C 男 26 C城市 12005000204 D 男 18 D城市 12005000205 E 女 21 E城市 12005000206 F 男 32 F城市 12005000207 G 女 29 G城市 12005000208 H 女 30 H城市 12005000209 I 男 26 I城市 12005000210 J 女 25 J城市

操作场景 HDFS集群可能出现DataNode节点间磁盘利用率不平衡的情况，比如集群中添加新数据节点的场景。如果HDFS出现数据不平衡的状况，可能导致多种问题，比如MapReduce应用程序无法很好地利用本地计算的优势、数据节点之间无法达到更好的网络带宽使用率或节点磁盘无法利用等等。所以MRS集群管理员需要定期检查并保持DataNode数据平衡。 HDFS提供了一个容量均衡程序Balancer。通过运行这个程序，可以使得HDFS集群达到一个平衡的状态，使各DataNode磁盘使用率与HDFS集群磁盘使用率的偏差不超过阈值。图1和图2分别是Balance前后DataNode的磁盘使用率变化。 图1 执行均衡操作前DataNode的磁盘使用率 图2 执行均衡操作后DataNode的磁盘使用率 均衡操作时间估算受两个因素影响： 需要迁移的总数据量： 每个DataNode节点的数据量应大于（平均使用率-阈值）*平均数据量，小于（平均使用率+阈值）*平均数据量。若实际数据量小于最小值或大于最大值即存在不平衡，系统选择所有DataNode节点中偏差最多的数据量作为迁移的总数据量。 Balancer的迁移是按迭代（iteration）方式串行顺序处理的，每个iteration迁移数据量不超过10GB，每个iteration重新计算使用率的情况。 因此针对集群情况，可以大概估算每个iteration耗费的时间（可以通过执行Balancer的日志观察到每次iteration的时间），并用总数据量除以10GB估算任务执行时间。 由于按iteration处理，Balancer可以随时启动或者停止。

相关文档 如果执行start-balancer.sh，“hadoop-root-balancer-主机名.out”日志显示“Access denied for user test1. Superuser privilege is required”，请参见执行balance常见问题定位方法。 如果在HDFS客户端启动一个Balance进程，该进程被异常停止后，再次执行Balance操作时失败，请参见HDFS执行Balance时被异常停止如何处理。 如果数据出现不均衡，某磁盘过满而其他磁盘未写满，请参见非HDFS数据残留导致数据分布不均衡。 如果单个节点内DataNode的各磁盘使用率不均匀，请参见节点内DataNode磁盘使用率不均衡。 如果需要平衡正在运行的单个DataNode上的磁盘数据，请参见配置HDFS DiskBalancer磁盘均衡。

配置描述 在Spark客户端的“{客户端安装目录}/Spark/spark/conf/core-site.xml”配置文件中修改配置。 表1 参数介绍 参数 描述 默认值 fs.obs.metrics.switch 上报OBS监控指标开关： true表示打开 false表示关闭 true fs.obs.metrics.consumer 指定OBS监控指标的处理方式。 org.apache.hadoop.fs.obs.metrics.OBSAMetricsProvider：表示收集统计OBS监控指标 org.apache.hadoop.fs.obs.DefaultMetricsConsumer：表示不收集OBS监控指标 要使用OBS监控功能，需确保上报OBS监控指标开关打开。 org.apache.hadoop.fs.obs.metrics.OBSAMetricsProvider

创建ZSTD_JNI压缩表 以Hive客户端安装用户登录安装客户端的节点。 执行以下命令，切换到客户端安装目录，例如安装目录为“/opt/client”，请用户根据实际情况修改。 cd /opt/client 执行以下命令配置环境变量。 source bigdata_env 集群是否启用了Kerberos认证。 是，执行以下命令进行用户认证，然后执行5。 kinit Hive业务用户 否，执行5。 执行以下命令登录Hive客户端。 beeline 创建ZSTD_JNI压缩格式的Hive表。 使用ZSTD_JNI压缩算法时，只需在创建ORC表时指定表属性参数“orc.compress”为“ZSTD_JNI”即可，例如： create table tab_1(...) stored as orc TBLPROPERTIES("orc.compress"="ZSTD_JNI"); ZSTD_JNI的压缩级别的取值范围为1~19，数值越高压缩比越高，相对压缩读写速率会变慢；数值越低压缩比越低，相对读写速率会变快，默认值为“6”。建表时设置表属性参数“orc.global.compress.level”即可设置压缩级别，如： create table tab_1(...) stored as orc TBLPROPERTIES("orc.compress"="ZSTD_JNI", 'orc.global.compress.level'='3'); 用户可以对特定的数据格式列指定压缩，以对业务数据进一步压缩。当前支持的特定格式数据包括：JSON数据列、BASE64数据列、时间戳数据列和UUID数据列。建表时设置表属性参数“orc.column.compress”即可。 例如，以下示例指定了压缩格式为ZSTD_JNI，压缩列f2为JSON格式的数据，f3为BASE64格式的数据，f4为时间戳格式的数据，f5为UUID格式的数据： create table test_orc_zstd_jni(f1 int, f2 string, f3 string, f4 string, f5 string) stored as orc TBLPROPERTIES('orc.compress'='ZSTD_JNI', 'orc.column.compress'='[{"type":"cjson","columns":"f2"},{"type":"base64","columns":"f3"},{"type ":"gorilla","columns":{"format": "yyyy-MM-dd HH:mm:ss.SSS", "columns": "f4"}},{"type":"uuid","columns":"f5"}]'); 用户可根据实际情况按照对应格式插入数据即可实现进一步压缩的效果。 执行以下命令查看表信息： desc formatted test; 回显结果中会显示压缩格式相关信息，例如图1表示Hive表的的压缩方式为“ZSTD_JNI”且压缩级别为“5”。 图1 查看表信息

操作步骤 可对INSERT...SELECT操作做如下的调优操作。 如果建的是Hive表，将存储类型设为Parquet，从而减少执行INSERT...SELECT语句的时间。 建议使用spark-sql或者在Beeline/JDBCServer模式下使用spark用户来执行INSERT...SELECT操作，避免执行更改文件owner的操作，从而减少执行INSERT...SELECT语句的时间。 在Beeline/JDBCServer模式下，executor的用户跟driver是一致的，driver是JDBCServer服务的一部分，是由spark用户启动的，因此其用户也是spark用户，且当前无法实现在运行时将Beeline端的用户透传到executor，因此使用非spark用户时需要对文件进行更改owner为Beeline端的用户，即实际用户。 如果查询的数据是大量的小文件将会产生大量map操作，从而导致输出存在大量的小文件，在执行重命名文件操作时将会耗费较多时间，此时可以通过设置“spark.sql.files.maxPartitionBytes”与“spark.files.openCostInBytes”来设置一个partition读取的最大字节，在一个partition中合并多个小文件来减少输出文件数及执行重命名文件操作的时间，从而减少执行INSERT...SELECT语句的时间。

相关文档 如果HDFS客户端写文件close失败，客户端提示数据块没有足够副本数，处理方法请参见HDFS客户端写文件close失败。 如果需要调整客户端日志级别，处理方法请参见调整HDFS Shell客户端日志级别。 如果使用HDFS客户端命令删除超长目录报错，处理方法请参见HDFS客户端无法删除超长目录。 如果集群外节点安装客户端使用hdfs命令上传文件失败，报错no route to host，处理方法请参见集群外节点安装客户端上传HDFS文件失败。 更多使用HDFS客户端常见问题如下：

如何设置HDFS客户端运行时的日志级别？ HDFS客户端运行时的日志是默认输出到管理控制台的，其级别默认为INFO。如果需要开启DEBUG级别日志，可以通过导出一个环境变量来设置，命令如下： export HADOOP_ROOT_ LOG GER=DEBUG,console 在执行完上面命令后，再执行HDFS Shell命令时，即可打印出DEBUG级别日志。 如果想恢复INFO级别日志，可执行如下命令： export HADOOP_ROOT_LOGGER=INFO,console

使用HDFS客户端 安装客户端，详细操作请参考使用MRS客户端。 以客户端安装用户，登录安装客户端的节点。 执行以下命令，切换到客户端安装目录，例如“/opt/client”。 cd /opt/client 执行以下命令配置环境变量。 source bigdata_env 如果集群为安全模式，执行以下命令进行用户认证。普通模式集群无需执行用户认证。 kinit 组件业务用户 执行HDFS Shell命令。例如： hdfs dfs -ls /

查询冻结表的冻结分区 查询冻结分区： show frozen partitions 表名; 默认元数据库冻结分区类型只支持int、string、varchar、date、timestamp类型。 外置元数据库只支持PostgreSQL数据库，且冻结分区类型只支持int、string、varchar、timestamp类型。 对冻结后的表进行Msck元数据修复时，需要先解冻数据。如果对冻结表进行过备份后恢复操作，则可以直接执行Msck元数据修复操作，且解冻只能通过msck repair命令进行操作。 对冻结后的分区进行rename时，需要先解冻数据，否则会提示分区不存在。 删除存在冻结数据的表时，被冻结的数据会同步删除。 删除存在冻结数据的分区时，被冻结的分区信息不会被删除，HDFS业务数据也不会被删除。 select查询数据时，会自动添加排查冷分区数据的过滤条件，查询结果将不包含冷分区的数据。 show partitions table查询表下的分区数据时，查询结果将不包含冷分区，可通过show frozen partitions table进行冷冻分区查询。

操作步骤 以Hive客户端安装用户登录源端集群安装客户端的节点。 执行以下命令，切换到客户端安装目录，例如安装目录为“/opt/client”，请用户根据实际情况修改。 cd /opt/client 执行以下命令配置环境变量。 source bigdata_env 如果集群开启了Kerberos认证，执行以下命令认证用户，否则跳过此步骤。 kinit Hive业务用户 执行以下命令登录Hive客户端。 beeline 执行以下命令创建数据库“dump_db”。 create database dump_db; 执行以下命令切换到“dump_db”数据库。 use dump_db; 执行以下命令在“dump_db”中创建表“test”。 create table test(id int); 执行以下命令向表“test”中插入数据。 insert into test values(123); 执行以下命令将数据库“dump_db”设置为复制策略的源。 alter database dump_db set dbproperties ('repl.source.for'='replpolicy1'); 执行alter命令修改数据库属性时，用户需要对该数据库拥有对应权限。权限设置方式如下： 集群未启用Ranger鉴权，需登录 FusionInsight Manager授予该用户所属角色管理员权限，详细操作请参考创建Hive用户并绑定角色章节。 集群启用了Ranger鉴权，需参考添加Hive的Ranger访问权限策略章节授予用户对应数据库的Repl Dump/Load操作权限。 删除设置了复制策略源的数据库时，需要先将该数据库的复制策略源设置为空，再对数据库执行删除操作，否则无法删除。将数据库复制策略源设置为空的命令如下： alter database dump_db set dbproperties ('repl.source.for'=''); 执行以下命令将“dump_db”导出到目标集群的“/user/hive/test”目录下。 repl dump dump_db with ('hive.repl.rootdir'='hdfs://haclusterX/user/hive/test'); “haclusterX”为新增的自定义参数“dfs.namenode.rpc-address.haclusterX”中的“haclusterX”。 指定导出目录时需要确保当前用户对该目录拥有读写权限。 以Hive客户端安装用户登录目标集群安装客户端的节点，并执行2-5。 执行以下命令将“/user/hive/test”目录下的“dump_db”数据库的数据导入到“load_db”数据库中。 repl load load_db from '/user/hive/repl'; 通过repl load导入数据库，指定数据库名称时需要注意以下情况： 指定的数据库不存在，在导入的过程中会创建对应的数据库。 指定的数据库已存在，且该数据库的“hive.repl.ckpt.key”属性值与导入的路径一致，则跳过导入操作。 指定的数据库已存在，但是该数据库下不存在任何表和functions，导入的过程中只将源数据库下的表导入到当前数据库中；如果该数据库下存在表或functions会导入失败。

操作示例 以Hive客户端安装用户登录安装客户端的节点。 执行以下命令，切换到客户端安装目录。 cd 客户端安装目录 例如安装目录为“/opt/client”，则执行以下命令： cd /opt/client 执行以下命令配置环境变量。 source bigdata_env 集群是否启用了Kerberos认证。 是，执行以下命令进行用户认证： kinit Hive业务用户 否，执行5。 执行以下命令登录Hive客户端。 beeline 执行如下命令创建Hive分区表“table_1”，指定分区“pt='2021-12-12'”的路径为“hdfs://xxx”，指定分区“pt='2021-12-18'”的路径为“obs://xxx”： 创建Hive分区表“table_1”： create table table_1(id string) partitioned by(pt string) [stored as [orc|textfile|parquet|...]]; 指定分区“pt='2021-12-12'”的路径为“hdfs://xxx”： alter table table_1 add partition(pt='2021-12-12') location 'hdfs://xxx'; 指定分区“pt='2021-12-12'”的路径为“obs://xxx”： alter table table_1 add partition(pt='2021-12-18') location 'obs://xxx'; 给分区表“table_1”中插入数据后，对应的分区数据存储在对应的存储源上，可以使用desc查看分区的Location，执行以下命令查看路径下的数据： desc formatted table_1 partition(pt='2021-12-18');

配置描述 可以通过以下两种方式配置是否过滤掉分区表分区路径不存在的分区。 在Spark客户端的“{客户端安装目录}/Spark/spark/conf/spark-defaults.conf”配置文件中进行设置，修改如下参数： 表1 参数说明 参数 参数说明 取值示例 spark.sql.hive.verifyPartitionPath 配置读取Hive分区表时，是否过滤掉分区表分区路径不存在的分区。 “true”：过滤掉分区路径不存在的分区。 “false”：不进行过滤。 true 在spark-submit命令提交应用时，通过“--conf”参数配置是否过滤掉分区表分区路径不存在的分区。 示例： spark-submit --class org.apache.spark.examples.SparkPi --conf spark.sql.hive.verifyPartitionPath=true $SPARK_HOME/lib/spark-examples_*.jar

约束与限制 不能对Hudi表映射的Hive外部表做增删改操作（即insert、update、delete、load、merge、alter、msck），只支持查询操作（select）。 表授权 不支持修改类权限操作（即update、Alter、write、All）。 备份与恢复 由于ro表和rt表均由同一个Hudi源表映射的，备份其中一张表，另一张也会跟着备份，恢复也是同时恢复的，因此只需备份其中一张表即可。 组件版本 Hive：FusionInsight_HD_xxx，Hive内核版本3.1.0。 Spark2x：FusionInsight_Spark2x_xxx，Hudi内核版本：0.11.0。

查询Hudi的Schema演进表对应的Hive外部表 如果该Hudi表为Schema演进表（表的字段执行过修改），则在Hive客户端查询该表时还需额外设置一个参数： set hive.exec.schema.evolution=true; 例如以cow表实时视图的查询举例，其他各个视图的查询都要额外添加该参数： 设置“hive.exec.schema.evolution”为“true”： set hive.exec.schema.evolution=true; 查询Hive外部表： select * from hudicow;

Hudi表对应的Hive外部表介绍 Hudi源表对应一份HDFS的数据，通过Spark组件、Flink组件或者Hudi客户端，可以将Hudi表的数据映射为Hive外部表，基于该外部表，Hive可以进行实时视图查询、读优化视图查询以及增量视图查询。 根据Hudi源表的类型的不同，提供不同的视图查询： Hudi源表类型为Copy On Write时，可以映射为Hive的一张外部表，该表可以提供实时视图查询以及增量视图查询。 Hudi源表类型为Merge On Read时，可以映射为Hive的两张外部表（ro表和rt表），ro表提供读优化视图查询，rt表提供实时视图查询以及增量视图查询。

配置参数 在Spark客户端的“{客户端安装目录}/Spark/spark/conf/spark-defaults.conf”配置文件中进行设置。 参数 说明 取值示例 spark.sql.streaming.stateStore.providerClass 用于管理有状态流查询中的状态数据的类。此类必须是StateStoreProvider的子类，并且必须具有零参数构造函数。 配置参数值为org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider即可选择RocksDB作为状态后端。 org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider

配置场景 状态信息存储在默认的HDFS BackedStateStore，随着状态数据的增多，垃圾回收器需要处理更多的对象，从而增加了垃圾回收的时间和开销，导致JVM GC占用大量时间。可以通过配置参数spark.sql.streaming.stateStore.providerClass，选择RocksDB作为状态后端。 RocksDB是一个嵌入式的键值存储引擎，它将数据存储在本地磁盘上，支持高效的读写操作。具有可定制的内存管理和压缩算法，能根据不同的应用场景进行优化。常用于存储大量的结构化或半结构化数据，能有效处理大规模的状态数据。

操作场景 对于Spark应用来说，资源是影响Spark应用执行效率的一个重要因素。当一个长期运行的服务（比如JDBCServer），如果分配给它多个Executor，可是却没有任何任务分配给它，而此时有其他的应用却资源紧张，这就造成了很大的资源浪费和资源不合理的调度。 动态资源调度就是为了解决这种场景，根据当前应用任务的负载情况，实时的增减Executor个数，从而实现动态分配资源，使整个Spark系统更加健康。

ClickHouse通过MySQL引擎对接RDS服务 MySQL引擎用于将远程的MySQL服务器中的表映射到ClickHouse中，并允许您对表进行INSERT和SELECT查询，以方便您在ClickHouse与MySQL之间进行数据交换。 MySQL引擎使用语法： CREATE DATABASE [IF NOT EXISTS] db_name [ON CLUSTER cluster] ENGINE = MySQL('host:port', ['database' | database], 'user', 'password') 表1 MySQL数据库引擎参数说明 参数 参数说明 host:port RDS服务MySQL数据库实例IP地址和端口。 database RDS服务MySQL数据库名。 user RDS服务MySQL数据库用户名。 password RDS服务MySQL数据库用户密码，命令中如果携带认证密码信息可能存在安全风险，在执行命令前建议关闭系统的history命令记录功能，避免信息泄露。 MySQL引擎使用示例： 连接到RDS服务的MySQL数据库。详细操作可以参考RDS服务MySQ L实例 连接。 在MySQL数据库上创建表，并插入数据。 创建表mysql_table： CREATE TABLE `mysql_table` ( `int_id` INT NOT NULL AUTO_INCREMENT, `float` FLOAT NOT NULL, PRIMARY KEY (`int_id`)); 插入表数据： insert into mysql_table (`int_id`, `float`) VALUES (1,2); 登录ClickHouse客户端安装节点。执行以下命令，切换到客户端安装目录。 cd /opt/client 执行以下命令配置环境变量。 source bigdata_env 如果当前集群已启用Kerberos认证，执行以下命令认证当前用户，当前用户需要具有创建ClickHouse表的权限，具体请参见创建ClickHouse角色章节，为用户绑定对应角色。如果当前集群未启用Kerberos认证，则无需执行本步骤。 如果是MRS 3.1.0版本集群，则需要先执行： export CLICKHOUSE_SECURITY_ENABLED=true 执行安全认证： kinit 组件业务用户 例如： kinit clickhouseuser 使用客户端命令连接ClickHouse。 clickhouse client --host clickhouse实例IP --user 用户名 --password --port 端口号 输入用户密码 在ClickHouse中创建MySQL引擎的数据库，创建成功后自动与MySQL服务器交换数据。 CREATE DATABASE mysql_db ENGINE = MySQL('RDS服务MySQL数据库实例IP地址:MySQL数据库实例端口', 'MySQL数据库名', 'MySQL数据库用户名', 'MySQL数据库用户名密码'); 切换到新建的数据库mysql_db，并查询表数据。 USE mysql_db; 在ClickHouse中查询MySQL数据库表数据。 SELECT * FROM mysql_table; 执行结果如下： ┌─int_id─┬─float─┐ │ 1 │ 2 │ └─────┴──── ┘ 新增插入数据后也可以正常进行查询。 INSERT INTO mysql_table VALUES (3,4); SELECT * FROM mysql_table; 执行结果如下： ┌─int_id─┬─float─┐ │ 1 │ 2 │ │ 3 │ 4 │ └─────┴──── ┘

约束与限制 只支持两表Join的场景。 不支持FULL OUTER JOIN的数据倾斜处理。 示例：执行下面SQL语句，a表倾斜或b表倾斜都无法触发该优化。 select aid FROM a FULL OUTER JOIN b ON aid=bid; 不支持LEFT OUTER JOIN的右表倾斜处理。 示例：执行下面SQL语句，b表倾斜无法触发该优化。 select aid FROM a LEFT OUTER JOIN b ON aid=bid; 不支持RIGHT OUTER JOIN的左表倾斜处理。 示例：执行下面SQL语句，a表倾斜无法触发该优化。 select aid FROM a RIGHT OUTER JOIN b ON aid=bid;

配置场景 在Spark SQL多表Join的场景下，会存在关联键严重倾斜的情况，导致Hash分桶后，部分桶中的数据远高于其他分桶。最终导致部分Task过重，运行很慢；其他Task过轻，运行很快。一方面，数据量大Task运行慢，使得计算性能低；另一方面，数据量少的Task在运行完成后，导致很多CPU空闲，造成CPU资源浪费。 通过如下配置项可开启自动进行数据倾斜处理功能，通过将Hash分桶后数据量很大的、且超过数据倾斜阈值的分桶拆散，变成多个task处理一个桶的数据机制，提高CPU资源利用率，提高系统性能。未产生倾斜的数据，将采用原有方式进行分桶并运行。

配置描述 在Spark客户端的“{客户端安装目录}/Spark/spark/conf/spark-defaults.conf”配置文件中添加如下表格中的参数。 表1 参数说明 参数 参数说明 取值示例 spark.sql.adaptive.enabled 用于启用或禁用自适应查询执行（Adaptive Query Execution，AQE）功能。该功能允许Spark在运行时动态调整查询计划，以应对数据倾斜、统计信息不准确等问题，从而提升查询性能。 注意：AQE特性与DPP（动态分区裁剪）特性同时开启时，SparkSQL任务执行中会优先执行DPP特性，从而使得AQE特性不生效。集群中DPP特性是默认开启的，因此开启AQE特性的同时，需要将DPP特性关闭。 false spark.sql.optimizer.dynamicPartitionPruning.enabled 用于控制动态分区裁剪（Dynamic Partition Pruning，DPP） 的启用状态。该功能允许Spark在查询执行时根据运行时数据动态过滤分区，从而减少数据扫描量，提升查询性能。 true spark.sql.adaptive.skewJoin.enabled 用于控制自适应倾斜Join优化的启用状态。该功能允许Spark在运行时检测并自动处理数据倾斜问题，避免因个别Task处理大量数据而导致的长尾性能问题。 当此配置为true且spark.sql.adaptive.enabled设置为true时，启用运行时自动处理join运算中的数据倾斜功能。 true spark.sql.adaptive.skewJoin.skewedPartitionFactor 此配置为一个倍数因子，用于判定分区是否为数据倾斜分区。单个分区被判定为数据倾斜分区的条件为：当一个分区的数据大小超过除此分区外其他所有分区大小的中值与该配置的乘积，并且大小超过spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes配置值时，此分区被判定为数据倾斜分区。 5 spark.sql.adaptive.skewjoin.skewedPartitionThresholdInBytes 分区大小（单位：字节）大于该阈值且大于spark.sql.adaptive.skewJoin.skewedPartitionFactor与分区中值的乘积，则认为该分区存在倾斜。理想情况下，此配置应大于spark.sql.adaptive.advisoryPartitionSizeInBytes。 256MB spark.sql.adaptive.shuffle.targetPostShuffleInputSize 每个Task处理的shuffle数据的最小数据量。单位：Byte。 67108864

操作场景 HDFS Colocation（同分布）是HDFS提供的数据分布控制功能，利用HDFS Colocation接口，可以将存在关联关系或者可能进行关联操作的数据存放在相同的存储节点上。Hive支持HDFS的Colocation功能，即在创建Hive表时，设置表文件分布的locator信息，当使用insert语句向该表中插入数据时会将该表的数据文件存放在相同的存储节点上（不支持其他数据导入方式），从而使后续的多表关联的数据计算更加方便和高效。支持HDFS的Colocation功能的Hive表格式只有TextFile和RCFile。

操作步骤 以Hive客户端安装用户登录安装客户端的节点。 执行以下命令，切换到客户端安装目录。 cd 客户端安装目录 例如安装目录为“/opt/client”，则执行以下命令： cd /opt/client 执行以下命令配置环境变量。 source bigdata_env 集群是否启用了Kerberos认证。 是，执行以下命令进行用户认证： kinit Hive业务用户 否，执行5。 执行以下命令，将需要关联的关系型数据库驱动Jar包上传到HDFS目录下。 hdfs dfs -put Jar包所在目录 保存Jar包的HDFS目录 例如将“/opt”目录下Oracle驱动Jar包上传到HDFS的“/tmp”目录下，则执行如下命令。 hdfs dfs -put /opt/ojdbc6.jar /tmp 按照如下示例，在Hive客户端创建关联关系型数据库的外表。以下为关联Oracle linux6版本示例： 如果集群开启了Kerberos认证，创建Hive表的用户需要“ADMIN”权限，并设置admin权限： set role admin; 添加连接关系型数据库的驱动Jar包，不同数据库有不同的驱动Jar，且Jar包所在路径请以实际路径为准： ADD JAR hdfs:///tmp/ojdbc6.jar; 执行以下命令创建Hive表，且表的列需比数据库返回结果多一列用于分页查询： CREATE EXTERNAL TABLE ora_test (id STRING,rownum string) STORED BY 'com.qubitproducts.hive.storage.jdbc.JdbcStorageHandler' TBLPROPERTIES ( "qubit.sql.database.type" = "ORACLE", "qubit.sql.jdbc.url" = "jdbc:oracle:thin:@//10.163.xxx.xxx:1521/mydb", "qubit.sql.jdbc.driver" = "oracle.jdbc.OracleDriver", "qubit.sql.query" = "select name from aaa", "qubit.sql.column.mapping" = "id=name", "qubit.sql.dbcp.username" = "test", "qubit.sql.dbcp.password" = "xxx"); 表1 创建Hive表参数说明 参数 参数说明 qubit.sql.database.type 用于设置关联的关系型数据库类型，例如“ORACLE”。 qubit.sql.jdbc.url 通过JDBC连接关系型数据库的URL，不同数据库有不同的URL格式，例如“jdbc:oracle:thin:@//10.163.xxx.xxx:1521/mydb”。 qubit.sql.jdbc.driver 用于配置关系型数据库驱动类名，例如“oracle.jdbc.OracleDriver”。 qubit.sql.query 在关系型数据库中执行的查询SQL语句，结果将返回至Hive表中。 qubit.sql.column.mapping 可选，Hive表的列与关系型数据库表的列进行匹配。 qubit.sql.dbcp.username 连接关系型数据库的用户名。 qubit.sql.dbcp.password 连接关系型数据库的用户密码。 命令中如果携带认证密码信息可能存在安全风险，在执行命令前建议关闭系统的history命令记录功能，避免信息泄露。