欢迎关注大数据技术架构与案例微信公众号:过往记忆大数据
过往记忆博客公众号iteblog_hadoop
欢迎关注微信公众号:
过往记忆大数据

Spark SQL 查询中 Coalesce 和 Repartition 暗示(Hint)

如果你使用 Spark RDD 或者 DataFrame 编写程序,我们可以通过 coalescerepartition 来修改程序的并行度:

val data = sc.newAPIHadoopFile(xxx).coalesce(2).map(xxxx)
或
val data = sc.newAPIHadoopFile(xxx).repartition(2).map(xxxx)

val df = spark.read.json("/user/iteblog/json").repartition(4).map(xxxx)
val df = spark.read.json("/user/iteblog/json").coalesce(4).map(xxxx)

通过 coalescerepartition 函数我们一方面可以减少 Task 数据从未达到减少作业输出文件的数量;同时我们也可以加大并行度从而提高程序的运行效率。

Spark SQL 查询中 Coalesce 和 Repartition 暗示(Hint)
如果想及时了解Spark、Hadoop或者HBase相关的文章,欢迎关注微信公众号:iteblog_hadoop

我们现在越来越多的人使用 Spark SQL 来编写程序,可是在 Spark 2.4 之前,我们是不能直接在 SQL 里面使用 coalescerepartition的。值得高兴的是,国内的开发者为 Spark SQL 开发了一个功能,使得我们在 Spark SQL 里面也能用这两个函数,详见 SPARK-24940。这个功能在 Spark 2.4 已经发布了,这样我们可以通过 COALESCEREPARTITION 关键字暗示来设置程序的并行度。使用如下:

package com.iteblog

import java.util.UUID

import org.apache.spark.sql.SparkSession

object Iteblog {

  case class Person(name: String, age: Int)

  def main(args: Array[String]) {

    val spark = SparkSession
      .builder()
      .appName("iteblog example")
      .master("local[2]")
      .enableHiveSupport()
      .getOrCreate()

    // For implicit conversions like converting RDDs to DataFrames
    import spark.implicits._

    val person = 1.to(10000).map { i =>
      Person(UUID.randomUUID().toString.substring(1, 6), i % 100)
    }

    val df = spark.sparkContext.parallelize(person,2).toDF()
    df.createOrReplaceTempView("person")
    spark.sql("create table iteblog0 as select age,count(*) from person where age between 10 and 20 group by age").explain()
  }
}

上面程序的物理计划如下:

== Physical Plan ==
Execute CreateHiveTableAsSelectCommand CreateHiveTableAsSelectCommand [Database:default}, TableName: iteblog0, InsertIntoHiveTable]
+- *(2) HashAggregate(keys=[age#4], functions=[count(1)])
   +- Exchange hashpartitioning(age#4, 200)
      +- *(1) HashAggregate(keys=[age#4], functions=[partial_count(1)])
         +- *(1) Project [age#4]
            +- *(1) Filter ((age#4 >= 10) && (age#4 <= 20))
               +- *(1) SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, qwe.App$Person, true]).name, true, false) AS name#3, assertnotnull(input[0, qwe.App$Person, true]).age AS age#4]
                  +- Scan[obj#2]

如果我们加上 REPARTITION 关键字暗示,如下:

spark.sql("create table iteblog1 as select /*+ REPARTITION(4) */ age,count(*) from person where age between 10 and 20 group by age").explain()

则物理计划变成下面的

== Physical Plan ==
Execute CreateHiveTableAsSelectCommand CreateHiveTableAsSelectCommand [Database:default}, TableName: iteblog1, InsertIntoHiveTable]
+- Exchange RoundRobinPartitioning(4)
   +- *(2) HashAggregate(keys=[age#4], functions=[count(1)])
      +- Exchange hashpartitioning(age#4, 200)
         +- *(1) HashAggregate(keys=[age#4], functions=[partial_count(1)])
            +- *(1) Project [age#4]
               +- *(1) Filter ((age#4 >= 10) && (age#4 <= 20))
                  +- *(1) SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, qwe.App$Person, true]).name, true, false) AS name#3, assertnotnull(input[0, qwe.App$Person, true]).age AS age#4]
                     +- Scan[obj#2]

可以看到第四行多了 +- Exchange RoundRobinPartitioning(4),其他的不变。通过指定 coalescerepartition 暗示,我们就可以在 Spark SQL 里面指定并行度。

注意,如果你使用 Spark 2.4 以下版本,在 Spark SQL 里面加入 /*+ REPARTITION(4) */ 暗示,语句也不会运行错误,只不过并不会修改如何并行度相关属性而已。

本博客文章除特别声明,全部都是原创!
原创文章版权归过往记忆大数据(过往记忆)所有,未经许可不得转载。
本文链接: 【Spark SQL 查询中 Coalesce 和 Repartition 暗示(Hint)】(https://www.iteblog.com/archives/2501.html)
喜欢 (12)
分享 (0)
发表我的评论
取消评论

表情
本博客评论系统带有自动识别垃圾评论功能,请写一些有意义的评论,谢谢!