Sparksql.sql.codegen 没有提供任何改进

我正在 Spark sql 中执行查询，如下所示。表的数据存储在 Hive 表的 2 个不同节点中。

但由于查询有点慢，我尝试在 Spark 中找到一些选项，以便查询可以执行得更快。所以我发现我们可以配置sparksql.sql.codegen and spark.sql.inMemoryColumnarStorage.compressed为 true 而不是默认的 false。

但我没有任何改进，这两个选项为 true 的查询需要 4.1 分钟来执行。使用此选项为 false 也需要 4.1 分钟。

您知道为什么这个选项不起作用吗？

   query = hiveContext.sql("""select
        l_returnflag,
        l_linestatus,
        sum(l_quantity) as sum_qty,
        sum(l_extendedprice) as sum_base_price,
        sum(l_extendedprice * (1 - l_discount)) as sum_disc_price,
        sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) as sum_charge,
        avg(l_quantity) as avg_qty,
        avg(l_extendedprice) as avg_price,
        avg(l_discount) as avg_disc,
        count(*) as count_order
    from
        lineitem
    where
        l_shipdate <= '1998-09-16'
    group by
        l_returnflag,
        l_linestatus
    order by
        l_returnflag,
        l_linestatus""");

query.collect();

• spark.sql.codegen.wholeStageSpark 2.0 默认启用。它会做所有的Spark Catalist 方面可以进行内部优化 .

• spark.sql.codegen（这是 Spark 1.3+ 中的功能）默认情况下false。即使你说的是真的，你也可以与DF.explain / debug

但是，请。重新访问 Spark 2+ 中解释的方法，如下所示。

如果您使用的是较低版本的 Spark，即 1.3 或 1.4+，则相同的 DataFrame 方法是有效的，除非我们必须与 hiveContext 一起使用。

• 根据我的经验，上述查询的 Dataset[Row] 又名 DataFrame 方法比普通的 hive 查询要快一些。

请尝试下面的伪代码。

创建一个没有任何聚合、分组、排序的数据框，如下所示。

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);

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