皮皮网

1.Blaze：SparkSQL Native 算子优化在快手的算k算设计与实践
2.Spark中cache和persist的区别
3.SPARK-38864 - Spark支持unpivot源码分析
4.Spark repartition和coalesce的区别
5.Spark原理详解

Blaze：SparkSQL Native 算子优化在快手的设计与实践

       Blaze：SparkSQL Native 算子优化在快手的设计与实践

       在当前时代，Spark因其相比Hive的源码强大性能，已成为众多公司主要的详解执行引擎。随着业务的算k算不断扩展和数据规模的提升，对Spark性能提升的源码追求从未停止。Spark性能提升主要集中在两个方向：执行计划优化和运行时执行效率优化。详解口袋任务源码

       分享将围绕Blaze项目展开，算k算探索其在快手的源码设计与实践。Blaze是详解一个基于Apache DataFusion项目封装的向量化执行引擎中间件，旨在充分利用Spark分布式计算框架的算k算优势，同时发挥DataFusion Native向量化算子执行的源码性能优势。

       首先，详解回顾Spark的算k算发展历程，从1.0版本的源码解释执行模型到2.0版本的多算子编译，再到3.0版本引入的详解Adaptive Query Execution（AQE）自适应执行引擎，直至未来的优化方向——向量化执行。

       接着，探讨向量化执行的进展，包括Velox、电脑游戏源码论坛Gluten、Photon和Native Codegen等公司的探索与实践，以及它们如何结合Spark生态进行性能提升。

       Blaze项目在快手的探索始于两年前，经过持续迭代，目前已具备上线使用的能力。 Blaze基于Apache DataFusion项目封装，通过扩展组件将Spark生成的物理执行计划转换为对应的Native执行计划，进而传递给底层的DataFusion执行引擎。

       整体架构包括Spark on Blaze架构的流向，展示如何通过Blaze Extension组件将Spark执行流程与DataFusion Native执行引擎连接起来，以及执行计划转换、Native计划生成与提交的过程。核心组件包括Blaze Session Extension、Plan SerDe、JNI Gateways和Native Operators。

       执行过程分为物理执行计划的转换、Native计划生成与提交，免费源码 涛涛博客以及Native执行三个阶段。物理执行计划转换部分涉及算子检查、翻译与优化策略，以确保转换对性能的影响降到最低。Native计划生成与提交涉及序列化与JNI传递，而Native执行则通过DataFusion框架异步执行。

       Blaze项目在实际应用中实现了对TPC-DS数据集的高效处理，支持大量业务UDF调用，并提供了内存管理、UDF兼容等优化。此外，项目还实现了对特定算子的深度定制优化，如排序、聚合等。

       当前进展包括算子覆盖度的提升、基准测试结果、线上收益及未来工作规划。算子覆盖度基本按照线上使用频率推进，玩呗斗牌游戏源码支持常见的算子，对某些算子进行优化和改进。性能测试显示在特定场景下能够实现高达倍的性能提升。线上应用已实现部分生产ETL任务的性能提升，单任务性能提升最多可达四倍以上。

       未来工作计划包括持续优化算子支持、大规模推广、接口抽象以支持更多引擎以及回馈开源社区。项目已开放初步可用版本，欢迎更多开发者参与共建。

       本文总结了Blaze项目在快手的设计与实践，展示了其在Spark性能提升方面的探索与应用，为业界提供了宝贵的参考与借鉴。

Spark中cache和persist的区别

       cache

       ã€€ã€€é»˜è®¤æ˜¯å°†æ•°æ®å­˜æ”¾åˆ°å†…存中，懒执行

       ã€€ã€€def cache(): this.type = persist()

       ã€€ã€€persist

       ã€€ã€€å¯ä»¥æŒ‡å®šæŒä¹…化的级别。

       ã€€ã€€æœ€å¸¸ç”¨çš„是MEMORY_ONLY和MEMORY_AND_DISK。

       ã€€ã€€â€_2”表示有副本数。尽量避免使用_2和DISK_ONLY级别

       ã€€ã€€cache和persist的注意点

       ã€€ã€€1.都是懒执行(有的叫延迟执行)，需要action触发执行，最小单位是partition

       ã€€ã€€2.对一个RDD进行cache或者persist之后，下次直接使用这个变量，就是使用持久化的数据

       ã€€ã€€3.如果使用第二种方式，不能紧跟action算子

SPARK- - Spark支持unpivot源码分析

       unpivot是数据库系统中用于列转行的内置函数，如SQL SERVER, Oracle等。以数据集tb1为例，每个数字代表某个人在某个学科的成绩。若要将此表扩展为三元组，mt4crm源码可使用union实现。但随列数增加，SQL语句变长。许多SQL引擎提供内置函数unpivot简化此过程。unpivot使用时需指定保留列、进行转行的列、新列名及值列名。

       SPARK从SPARK-版本开始支持DataSet的unpivot函数，逐步扩展至pyspark与SQL。在Dataset API中，ids为要保留的Column数组，Column类提供了从String构造Column的隐式转换，方便使用。利用此API，可通过unpivot函数将数据集转换为所需的三元组。values表示转行列，variableColumnName为新列名，valueColumnName为值列名。

       Analyser阶段解析unpivot算子，将逻辑执行计划转化为物理执行计划。当用户开启hive catalog，SPARK SQL根据表名和metastore URL查找表元数据，转化为Hive相关逻辑执行计划。物理执行计划如BroadcastHashJoinExec，表示具体的执行策略。规则ResolveUnpivot将包含unpivot的算子转换为Expand算子，在物理执行计划阶段执行。此转换由开发者自定义规则完成，通过遍历逻辑执行计划树，根据节点类型及状态进行不同处理。

       unpivot函数实现过程中，首先将原始数据集投影为包含ids、variableColumnName、valueColumnName的列，实现语义转换。随后，通过map函数处理values列，构建新的行数据，最终返回Expand算子。在物理执行计划阶段，Expand算子将数据转换为所需形式，实现unpivot功能。

       综上所述，SPARK内置函数unpivot的实现通过解析列参数，组装Expand算子完成，为用户提供简便的列转行功能。通过理解此过程，可深入掌握SPARK SQL的开发原理与内在机制。

Spark repartition和coalesce的区别

       æœ‰äº›æ—¶å€™ï¼Œåœ¨å¾ˆå¤špartition的时候，我们想减少点partition的数量，不然写到HDFS上的文件数量也会很多很多。

        我们使用reparation呢，还是coalesce。所以我们得了解这两个算子的内在区别。

        要知道，repartition是一个消耗比较昂贵的操作算子，Spark出了一个优化版的repartition叫做coalesce，它可以尽量避免数据迁移，

        但是你只能减少RDD的partition.

        举个例子，有如下数据节点分布：

        用coalesce，将partition减少到2个：

        注意，Node1 和 Node3 不需要移动原始的数据

        The repartition algorithm does a full shuffle and creates new partitions with data that’s distributed evenly.

        Let’s create a DataFrame with the numbers from 1 to .

        repartition 算法会做一个full shuffle然后均匀分布地创建新的partition。我们创建一个1-数字的DataFrame测试一下。

        刚开始数据是这样分布的：

        我们做一个full shuffle，将其repartition为2个。

        这是在我机器上数据分布的情况：

        Partition A: 1, 3, 4, 6, 7, 9, ,

        Partition B: 2, 5, 8,

        The repartition method makes new partitions and evenly distributes the data in the new partitions (the data distribution is more even for larger data sets).

        repartition方法让新的partition均匀地分布了数据（数据量大的情况下其实会更均匀）

        coalesce用已有的partition去尽量减少数据shuffle。

        repartition创建新的partition并且使用 full shuffle。

        coalesce会使得每个partition不同数量的数据分布（有些时候各个partition会有不同的size）

        然而，repartition使得每个partition的数据大小都粗略地相等。

        coalesce 与 repartition的区别（我们下面说的coalesce都默认shuffle参数为false的情况）

        repartition(numPartitions:Int):RDD[T]和coalesce(numPartitions:Int，shuffle:Boolean=false):RDD[T] repartition只是coalesce接口中shuffle为true的实现

        有1w的小文件，资源也为--executor-memory 2g --executor-cores 2 --num-executors 5。

        repartition(4)：产生shuffle。这时会启动5个executor像之前介绍的那样依次读取1w个分区的文件，然后按照某个规则%4,写到4个文件中，这样分区的4个文件基本毫无规律，比较均匀。

        coalesce(4):这个coalesce不会产生shuffle。那启动5个executor在不发生shuffle的时候是如何生成4个文件呢，其实会有1个或2个或3个甚至更多的executor在空跑（具体几个executor空跑与spark调度有关，与数据本地性有关，与spark集群负载有关），他并没有读取任何数据！

        1.如果结果产生的文件数要比源RDD partition少，用coalesce是实现不了的，例如有4个小文件（4个partition），你要生成5个文件用coalesce实现不了，也就是说不产生shuffle，无法实现文件数变多。

        2.如果你只有1个executor（1个core），源RDD partition有5个，你要用coalesce产生2个文件。那么他是预分partition到executor上的，例如0-2号分区在先executor上执行完毕，3-4号分区再次在同一个executor执行。其实都是同一个executor但是前后要串行读不同数据。与用repartition(2)在读partition上有较大不同（串行依次读0-4号partition 做%2处理）。

        T表有G数据 有个partition 资源也为--executor-memory 2g --executor-cores 2 --num-executors 5。我们想要结果文件只有一个

Spark原理详解

       Spark原理详解：

       Spark是一个专为大规模数据处理设计的内存计算框架，其高效得益于其核心组件——弹性数据分布集RDD。RDD是Spark的数据结构，它将数据存储在分布式内存中，通过逻辑上的集中管理和物理上的分布式存储，提供了高效并行计算的能力。

       RDD的五个关键特性如下：

       每个RDD由多个partition组成，用户可以指定分区数量，默认为CPU核心数。每个partition独立处理，便于并行计算。

       Spark的计算基于partition，算子作用于partition上，无需保存中间结果，提高效率。

       RDD之间有依赖性，数据丢失时仅重新计算丢失分区，避免全量重算。

       对于key-value格式的RDD，有Partitioner决定分片和数据分布，优化数据处理的本地化。

       Spark根据数据位置调度任务，实现“移动计算”而非数据。

       Spark区分窄依赖（一对一）和宽依赖（一对多），前者不涉及shuffle，后者则会根据key进行数据切分。

       Spark的执行流程包括用户提交任务、生成DAG、划分stage和task、在worker节点执行计算等步骤。创建RDD的方式多样，包括程序中的集合、本地文件、HDFS、数据库、NoSQL和数据流等。

       技术栈方面，Spark与HDFS、YARN、MR、Hive等紧密集成，提供SparkCore、SparkSQL、SparkStreaming等扩展功能。

       在编写Spark代码时，首先创建SparkConf和SparkContext，然后操作RDD进行转换和应用Action，最后关闭SparkContext。理解底层机制有助于优化资源使用，如HDFS文件的split与partition关系。

       搭建Spark集群涉及上传、配置worker和master信息，以及启动和访问。内存管理则需注意Executor的off-heap和heap，以及Spark内存的分配和使用。