承接上文，本文演示如何控制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是同一个东西，一个分区对应一个文件。

想要了解跟多关于大数据培训课程内容欢迎关注尚硅谷大数据培训，尚硅谷除了这些技术文章外还有免费的高质量大数据培训课程视频供广大学员下载学习。

上一篇: Spark SQL编程DataFrame 创建_大数据培训
下一篇: SQL风格语法(主要)_大数据培训