玩转Spark Sql优化之提交参数控制(三)

承接上文，本文演示如何控制Spark Sql任务参数。

仍然是第一篇所提的三张表分表对应课程表、购物车表、支付表，三张表测试数据量分别为课程表3MB，购物车表4.3G，支付表2.3G。

小文件过多场景

Spark sql默认shuffle分区个数为200，参数由spark.sql.shuffle.partitions控制，此参数只能控制Spark sql、DataFrame、DataSet分区个数。不能控制RDD分区个数

所以如果两表进行join产生shuffle形成一张新表，如果新表的分区不进行缩小分区操作，那么就会有200份文件插入到hdfs上，这样就有可能导致小文件过多的问题。

还是由上面视图三张表为例，进行join，先不进行缩小分区操作。查看效果。为了演示效果，先禁用了广播join。广播join后面会进行说明。

import org.apache.spark.SparkConf
import org.apache.spark.sql.{SaveMode, SparkSession
object PartitionTuning {
      def main(args: Array[String]): Unit = {
           val sparkConf = new SparkConf().setAppName("test").set("spark.sql.autoBroadcastJoinThreshold","-1")
           val sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()
           val ssc = sparkSession.sparkContext           
testJoin(sparkSession)
      }                
      
def testJoin(sparkSession: SparkSession) = {
         //查询出三张表 并进行join 插入到最终表中
           val saleCourse = sparkSession.sql("select *from dwd.dwd_sale_course")          
           val coursePay = sparkSession.sql("select * from dwd.dwd_course_pay")           
               .withColumnRenamed("discount", "pay_discount")              
               .withColumnRenamed("createtime", "pay_createtime")
                              
           val courseShoppingCart = sparkSession.sql("select *from dwd.dwd_course_shopping_cart")          
                 .drop("coursename")
                 .withColumnRenamed("discount", "cart_discount")                 
                 .withColumnRenamed("createtime", "cart_createtime") 
                              
           saleCourse.join(courseShoppingCart, Seq("courseid", "dt", "dn"), "right")           
               .join(coursePay, Seq("orderid", "dt", "dn"), "left")                     
               .select("courseid", "coursename", "status", "pointlistid", "majorid", "chapterid",               
               "chaptername", "edusubjectid", "edusubjectname", "teacherid", "teachername", "coursemanager", "money", "orderid",               
               "cart_discount", "sellmoney","cart_createtime", "pay_discount", "paymoney", "pay_createtime", "dt", "dn")               
               .write.mode(SaveMode.Overwrite).insertInto("dws.dws_salecourse_detail")                
         }
 }            

提交yarn任务查看Spark Ui界面，对应200个分区（task）

查看HDFS上落盘的数据块，产生了200个文件

解决小文件过多问题也非常简单，在spark当中一个分区最终落盘形成一个文件，那么解决小文件过多问题只需将分区缩小即可。在插入表前，添加coalesce算子指定缩小后的分区个数。那么使用此算子需要注意,coalesce算子缩小分区后那么实际处理插入数据的任务只有一个，可能会导致oom，所以需要适当控制，并且coalesce算子里的参数只能填写比原有数据分区小的值，比如当前表的分区是200，那么填写参数必须小于200，否则无效。当然缩小分区后任务的耗时肯定会变久。

添加完coalesce算子后再次运行yarn任务，查看效果

最终产生的文件个数为20个，那么在Spark任务当中解决小文件过多的方案就是缩小分区个数。

提交参数控制

再次回到没有缩小分区之前的Stage当中

点击Stage查看task运行详情

可以看到task的分布并不均匀,vcore没有充分利用起来

根据当前任务的提交命令

spark-submit --master yarn --deploy-mode client --driver-memory 1g --num-executors 3 --executor-cores 4 --executor-memory 2g --queue spark --class com.atguigu.sparksqltuning.PartitionTuning spark-sql-tuning-1.0-SNAPSHOT-jar-with-dependencies.jar

去向yarn申请的executor vcore资源个数为12个（num-executors*executor-cores）,如果不修改spark sql分区个数，那么就会像上图所展示存在cpu空转的情况。这个时候需要合理控制shuffle分区个数。如果想要让任务运行得最快当然是一个task对应一个vcore,但是离线任务一般不会这样设置，为了合理利用资源，一般会将分区（也就是task）设置成vcore的2倍到3倍。

修改参数spark.sql.shuffle.partitions,此参数默认值为200。

那么根据我们当前任务的提交参数，将此参数设置为24或36为最优效果。

设置完参数，yarn上提交任务，再次运行

查看spark ui,点击相应stage,查看task详情

这张图就很明显了，分别hadoop101,hadop102,hadoop103各自申请到4个vcore,然后每个vcore都分配到了3个任务，也都是差不多时间点结束。充分利用了cpu的资源。

那么spark sql当中修改分区的方式就有3种了，分别是算子coalesce、repartition和参数spark.sql.shuffle.partitions

最终Stage id为4的join阶段，耗时也从3.3分钟降到了1.6分钟，优化效果非常明显。另一个join阶段也优化了一半（当时没截图）

结论

跑离线任务时我们可以合理控制分区数来提高效率，可以将分区数设置为executor一共申请vcore数的2倍或3倍。Spak Sql当中改变分区的方式有repartition、coalesce算子和spark.sql.shuffle.partitions参数，并且分区和task是同一个东西，一个分区对应一个文件。

大数据技术生态体系

大数据的切片机制有哪些

大数据logstsh架构

大数据技术kafka的零拷贝

大数据学习之部署Hadoop