spark3.0中如何实现查询计划

这篇文章主要介绍了spark 3.0中如何实现查询计划,具有一定借鉴价值,感兴趣的朋友可以参考下,希望大家阅读完这篇文章之后大有收获,下面让小编带着大家一起了解一下。

成都创新互联专注于企业营销型网站建设、网站重做改版、定陶网站定制设计、自适应品牌网站建设、HTML5商城开发、集团公司官网建设、成都外贸网站制作、高端网站制作、响应式网页设计等建站业务,价格优惠性价比高,为定陶等各大城市提供网站开发制作服务。

基本例子

我们考虑一个简单的例子,一个查询中涉及到filter以及aggregation,join操作的语句:

# in PySpark API:
query  = (
    questionsDF
    .filter(col('year') == 2019)
    .groupBy('user_id')
    .agg(
        count('*').alias('cnt')
    )
    .join(usersDF, 'user_id')
)

我们把例子中的usersDF是一组问问题的用户,这些问题用questionsDF来表示。这些问题用year的这一列来进行分区,代表着哪一年问的问题。在这个查询里,我们对2019年问问题的用户感兴趣,并且想知道每个人问了多少问题,而且我们想知道在输出中我们想知道一些额外信息,这就是为什么我们在聚合之后进行了usersDF的join操作。
这里有两种基本的方式去查看物理计划。第一种是在DataFrame上调用explain函数,该函数展现这个计划的文本化的展示: spark 3.0中如何实现查询计划

这在spark 3.0有了一些优化,explain函数带有了一个新参数 mode,这个参数的值可以是:formatted,cost,codegen。使用formatted模式将会把查询计划转化为更加有组织的输出(这里之展现了一部分): spark 3.0中如何实现查询计划 在formatted计划中,我们能看到裸数,改裸数只是展现了操作的名字并带有一个括号的数字。在数的下面,这里有一些数字对应的细节描述。cost模式将会展示除了物理计划之外的优化的逻辑计划,这些逻辑计划带有每个操作的统计信息,所以我们能看到在不同执行阶段的数据大小。最终codegen模式展现了将会执行的生成的java代码。
第二种方式是查看spark ui中的sql tab,这里有正在跑的和已经完成了的查询。通过点击你要查看的查询,我们可以看到物理计划的文本表示。在下面这个图片中,我们结合图形表示,文本表示以及它们之间的对应关系: spark 3.0中如何实现查询计划 不同点是图形表示的叶子节点在上面,根节点在下面,而文本表示的是反过来的。

CollapseCodegenStages

在物理计划的图形表示中,你能看到一些操作被组织成了一大块蓝色的矩形。这些大矩形对应着codegen阶段。这是发生在物理计划的优化阶段。这个是叫做CollapseCodegenStages来负责优化的,原理是把支持代码生成的操作聚合到一起,通过消除虚拟函数的调用来加速。但是并不是所有的操作支持代码生成。所以一些操作(如exchange操作)并不是大矩形的一部分。在我们的例子中,这里有三个codegen stages,对应着三个大矩形,你能在操作的括号中看到codegen stage的id。从这个树我们也可以分辨出一个操作是够支持代码生成,因为加入支持代码生成的话,这里将会在对应的操作的括号里有个星号。 spark 3.0中如何实现查询计划

我们简单的分析一下在我们查询中的每一个操作。

Scan parquet

scan parquet操作代表着从parquet文件中读取数据。从明细信息中,我们能直接看到从这个数据源中我们选择了哪些列。虽然我们没指定具体的字段,但是这里也会应用ColumnPruning规则,这个规则会确保只有真正字段才会从这个数据源中提取出来。我们也能看到有两种filters:PartitionFilters和PushFilters。PartitionFilters应用在数据源分区的字段上。这是非常重要的因为我们能跳过我们不需要的数据。检查对应的filters是否传播到正确的位置总是没错的。这是因为我们尽可能读取少量的数据,因为IO是比较费时的。在spark 2.4,这里还有一个代表实际读取到的分区的partitionCount字段,这个字段在spark 3.0已经去掉了。
PushFilters把字段直接下推到parquet文件中去,假如parquet文件过滤的列是按照过滤字段排序的话,这个规则就很有用了,因为这种情况下,我们能利用parquet内部结构去过滤数据。parquet文件是按照行组和每个行组的元数据文件组成的。这个元数据包含了每个行组的最大最小值,基于这个信息,我们就能判断是否读取这个行组。

Filter

Filter操作佷容易理解。它仅仅是代表过滤条件。但是这个操作怎么创建的并不是很明显,因为在查询中它并不是直接对应着过滤条件。因为所有的filters首先被Catalyst optimzer处理,改规则可能修改或者重新移动她们。这里有好几个规则在她们转换为物理计划前的逻辑计划。我们列举了一下:

  • PushDownPredicates-这个规则通过其他的操作把filter下推到离数据源更近的地方,但不是所有的操作都支持。比如,如果表达式不是确定性的,这就不行,假如我们使用类似first,last,collect_set,collect_list,rand等,filters操作就不能通过这些操作而进行下推,因为这些函数是不确定性的。

  • CombineFilters-结合两个临近的操作合成一个(收集两个filters条件合成一个更为复杂的的条件)

  • InferFiltersFromConstraints-这个规则实际上会创建新的filter操作,如从join操作(从inner join中创建一个joining key is not null)

  • PruneFilters-移除多余的filters(比如一个filters总是true)

Exchange

Exchange操作代表着shuffle操作,意味着物理数据的集群范围内的移动。这个操作是很费时的,因为它会通过网络移动数据。查询计划的信息也包含了一些数据重新分区的细节。在我们的例子中,是hashPartitioning(user_id,200): spark 3.0中如何实现查询计划 这意味着数据将会根据user_id列重新分区为200个分区,有着同样user_id的行将会属于同一个分区,将会分配到同一个executor上。为了确保只有200分区,spark将会计算user_id的hashcode并且对200取模。这个结果就是不同的user_ids就会分到同一个分区。同时有些分区可能是空的。这里也有其他类型的分区值的去留意一下:

  • RoundRobinPartitioning-数据将会随机分配到n个分区中,n在函数repartition(n)中指定

  • SinglePartition-所有数据将会分配到一个分区中,进而到一个executor中。

  • RangePartitioning-这个用在对数据排序中,用在orderBy或者sort操作中

HashAggregate

这个代表着数据聚合,这个经常是两个操作,要么被Exchange分开或者不分开: spark 3.0中如何实现查询计划 为什么这里有两个HashAggregate操作的原因是第一个是部分聚合,它在每个executor上每个分区分别进行聚合。在我们的例子中,你能看到partial_count(1)的function字段,最终的部分聚合结果就是第二个聚合。这个操作也展示了数据按照哪个分组的Keys字段。results字段展示了在聚合以后的可用的列。

BroadcastHashJoin & BroadcastExchange

BroadcastHashJoin(BHJ)代表着join算法的操作,除了这个,还有SortMergeJoin和ShuffleHashJoin。BHJ总是伴随着BroadcastExchange,这个代表着广播shuffle-数据将会收集到driver端并且会被传播到需要的executor上。

ColumnarToRow

这是在spark 3.0引入的新操作,用于列行之间的转换

感谢你能够认真阅读完这篇文章,希望小编分享的“spark 3.0中如何实现查询计划”这篇文章对大家有帮助,同时也希望大家多多支持创新互联,关注创新互联行业资讯频道,更多相关知识等着你来学习!


本文标题:spark3.0中如何实现查询计划
分享路径:http://scyanting.com/article/gpgpod.html