Spark Scala：多次查询同一个表

Question

我试图从同一个表（bigTable）查询多个列来生成一些聚合列（column1_sum，column2_sum，column3_count）。最后，我将所有列连接在一起形成一个表。

代码如下

val t1 = bigTable
            .filter($"column10" === value1)
            .groupBy("key1","key2")
            .agg(sum("column1") as "column1_sum")

val t2 = bigTable
            .filter($"column11"===1)
            .filter($"column10" === value1)
            .groupBy("key1","key2")
            .agg(sum("column2") as "column2_sum")

val t3 = bigTable
            .filter($"column10" === value3)
            .groupBy("key1","key2")
            .agg(countDistinct("column3") as "column3_count")

tAll
            .join(t1,Seq("key1","key2"),"left_outer")
            .join(t2,Seq("key1","key2"),"left_outer")
            .join(t3,Seq("key1","key2"),"left_outer")

上述代码的问题

bigTable是一个巨大的表（它可以运行数百万行）。因此，多次查询它并不高效。该查询需要花费大量时间才能运行。

关于如何以更有效的方式实现相同输出的任何想法？有没有办法查询bigTable次数较少？

非常感谢提前。

Answer 1

最简单的改进是仅执行单个聚合，其中谓词被推入CASE ... WHEN ...块，并用近似的等价替换countDistinct

tAll
  .groupBy("key1","key2")
  .agg(
    sum(
      when($"column10" === "value1", $"column1")
    ).as("column1_sum"),
    sum(
      when($"column10" === "value1" and $"column11" === 1, $"column2")
    ).as("column2_sum"),
    approx_count_distinct(
      when($"column10" === "value3", $"column3")
    ).as("column3_count"))
  .join(tAll, Seq("key1", "key2"), "right_outer"))

根据所使用的函数和有关数据分布的先验知识，您还可以尝试使用具有类似qazxsw poi逻辑的窗口函数替换聚合

CASE ... WHEN ...

但它通常是一种不太稳定的方法。

您还应该考虑使用分组列来存储

import org.apache.spark.sql.expressions.Window

val w = Window
 .partitionBy("key1", "key2")
 .rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)

tAll
  .withColumn(
    "column1_sum", 
    sum(when($"column10" === "value1", $"column1")).over(w))
 ...

：

bigTable

Answer 2

我的代码的一个主要改进是查询bigTable一次，而不是问题中提到的多次。

我正在尝试的一段代码（我的代码类似，这只是一个例子）：

val n: Int = ???  // Number of buckets
bigTable.write.bucketBy(n, "key1", "key2").saveAsTable("big_table_clustered")

val bigTableClustered = spark.table("big_table_clustered")