对于spark 2.0，默认情况下启用spark.sql.codegen.whitestage。它将做所有的内部优化可能从Spark催化剂方面。

spark.sql.codegen（这是spark 1.3+中的特性）默认为false。即使设置为true，也可以使用df.explain/debug进行交叉检查

但是，请。重新访问spark 2+中解释的方法，如下所示。
如果您使用的是spark的较低版本，即1.3或1.4+，则相同的Dataframe方法是有效的，除非我们必须与hivecontext一起使用。
根据我的经验，上述查询的dataset[row]aka dataframe方法比普通的hive查询快一些。
请尝试下面的伪代码。
创建一个不包含任何聚合、分组、排序依据的Dataframe。

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import spark.implicits._
import spark.sql

// warehouseLocation points to the default location for managed databases and tables
val warehouseLocation = "file:${system:user.dir}/spark-warehouse"

val spark = SparkSession
  .builder()
  .appName("Spark Hive Aggregations")
  .config("spark.sql.warehouse.dir", warehouseLocation)
  .enableHiveSupport()
  .getOrCreate()

val df : DataFrame = sql(("""select l_returnflag, l_linestatus,l_quantity,l_extendedprice,l_quantity ,l_extendedprice,l_quantity, l_extendedprice, l_discount from
        lineitem where l_shipdate <= '1998-09-16""");

// can use spark udf or when(cond, evaluation), instead of direct expression
 val df1 =  df.withColumn("sum_disc_price", df.col("l_extendedprice") * (1 - df.col("l_discount"))
          .withColumn("sum_charge", df.col("l_extendedprice") * (1 + df.col("l_tax"))

//NOW SUM, AVG and group by  on dataframe
val groupeddf = df1.groupBy(
  df1.col("returnflag")
, df1.col("l_linestatus")
.agg(
      avg(df1.col("l_quantity")),
    , avg(df1.col("l_extendedprice"))
    , avg(df1.col("l_discount"))
    , sum(df1.col("l_quantity"))
    , sum(df1.col("l_extendedprice"))
    , sum(df1.col("sum_disc_price"))
    , sum(df1.col("sum_charge"))
    , count(df1.col("l_linestatus").as("cnt")
    ) //end agg
    ) //end group by 
//order by on dataframe  
.orderBy("l_returnflag"))
.sort("l_linestatus")
val finalDF = groupeddf.select("l_returnflag","l_linestatus",............. etc);

此外，还需要考虑执行器内存、执行器/核心数等参数，以找到确切的问题