Spark CBO统计元数据

原创

Yiwenwu

修改于 2024-04-28 09:21:58

1941

修改于 2024-04-28 09:21:58

文章被收录于专栏：大数据&分布式大数据&分布式

解析流程

Spark SQL解析流程概述为：

SQL语句基于ANTLR4编译解析成AST树，SparkSqlParser#parse通过Visitor访问者模式遍历解析AST树，生成Unresolved Logical Plan(未解析逻辑计划)；
基于Analyzer#apply规则的匹配作用，绑定树节点信息(元数据Catalog)后生成Logical Plan(逻辑计划)；
基于Optimizer#apply优化低效的逻辑树结构(如CBO优化)，生成Optimized Logical Plan(优化逻辑计划)；
基于SparkPlanner#plan，根据Optimized Logical Plan匹配对应的Strategy并生成一组Physical Plans(物理计划)；
基于代价模型(Cost Model)选择出Selected Physical Plan(最优物理计划)并进行提交准备(prepareForExecution)；
最后基于代码生成获取执行RDD对象。

以上整个转换过程都在Spark的Driver端进行，不涉及分布式环境。

统计信息

Spark 2.2 开始支持CBO优化，触发统计元数据更新的时机如下：

ANALYZE：AnalyzeTableCommand、AnalyzeColumnCommand；
ALTER：AlterTableAddPartitionCommand、AlterTableDropPartitionCommand、AlterTableSetLocationCommand、TruncateTableCommand；
INSERT：InsertIntoHiveTable、InsertIntoHadoopFsRelationCommand、LoadDataCommand；

Statistics 统计信息，参考：org.apache.spark.sql.catalyst.plans.logical.Statistics

字段	字段名称
sizeInBytes	数据文件大小，单位byte
rowCount	表行数
attributeStats	字段属性Map

ColumnStat字段统计信息，字段支持直方图(Histograms)统计：

字段	字段名称
distinctCount	不同字段值的个数统计
min	字段最小值
max	字段最大值
nullCount	字段为null值的数量
avgLen	字段值平均长度
maxLen	字段值最大长度
histogram	字段值直方图
version	字段统计版本

Spark的元数据统计信息的获取有三种方式：

基于持久化的元数据metastore获取，目前仅支持Hive metastore；
基于InMemoryFileIndex，调用底层存储API(Hadoop API)计算数据文件的个数和存储大小；
使用Spark默认设置的数据大小，配置参数：spark.sql.defaultSizeInBytes；

对接外部元数据metastore封装为CatalogStatistics，表的元数据信息从Table#parameters获取，统计信息存储和读取封装分别调用：HiveExternalCatalog#statsToProperties、HiveExternalCatalog#statsFromProperties，HiveClient封装基于Hive Metastore RPC接口，通过调用HiveClient读写保存在Hive Metastore的统计元数据。Spark表统计有如下配置主键：

spark.sql.statistics.totalSize：表数据文件总大小，单位byte；
spark.sql.statistics.numRows：表数据总行数；
spark.sql.statistics.colStats.{字段名}：其中value值是统计json，包括信息有 distinctCount, min, max, nullCount, avgLen, maxLen, histogram, version；

ANALYZE执行

ANALYZE TABLE table_name [ PARTITION ( partition_col_name [ = partition_col_val ] [ , ... ] ) ]
    COMPUTE STATISTICS [ NOSCAN | FOR COLUMNS col [ , ... ] | FOR ALL COLUMNS ]

Analyze命令支持：AnalyzeColumnCommand、AnalyzeTableCommand

统计元数据执行：

表统计信息：调用CommandUtils#calculateTotalSize?封装执行，totalSize读取Location下FS文件信息并汇总，numRows调用spark运行程序sparkSession.table("table").count()；
字段统计信息：调用CommandUtils#computeColumnStats 封装执行，CommandUtils#computePercentiles 支持直方图统计各分段数据(默认254分段)，字段统计执行Spark?Aggregate算子实现；

统计元数据获取：获取表、分区信息会自动补充上统计元数据信息；

统计元数据更新：SessionCatalog#alterTableStats更新元数据，针对Hive Metastore实现，则调用HiveClient#alterTable方法更新元数据配置信息。

我正在参与2024腾讯技术创作特训营最新征文，快来和我瓜分大奖！

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

大数据处理套件 TBDS