我在 TPCDS 基准的查询 47 中第一次遇到这种行为。
为澄清起见，这是查询。
--q47.sql--
with v1 as(
select i_category, i_brand,
s_store_name, s_company_name,
d_year, d_moy,
sum(ss_sales_price) sum_sales,
avg(sum(ss_sales_price)) over
(partition by i_category, i_brand,
s_store_name, s_company_name, d_year)
avg_monthly_sales,
rank() over
(partition by i_category, i_brand,
s_store_name, s_company_name
order by d_year, d_moy) rn
from item, store_sales, date_dim, store
where ss_item_sk = i_item_sk and
ss_sold_date_sk = d_date_sk and
ss_store_sk = s_store_sk and
(
d_year = 1999 or
( d_year = 1999-1 and d_moy =12) or
( d_year = 1999+1 and d_moy =1)
)
group by i_category, i_brand,
s_store_name, s_company_name,
d_year, d_moy),
v2 as(
select v1.i_category, v1.i_brand, v1.s_store_name, v1.s_company_name, v1.d_year,
v1.d_moy, v1.avg_monthly_sales ,v1.sum_sales, v1_lag.sum_sales psum,
v1_lead.sum_sales nsum
from v1, v1 v1_lag, v1 v1_lead
where v1.i_category = v1_lag.i_category and
v1.i_category = v1_lead.i_category and
v1.i_brand = v1_lag.i_brand and
v1.i_brand = v1_lead.i_brand and
v1.s_store_name = v1_lag.s_store_name and
v1.s_store_name = v1_lead.s_store_name and
v1.s_company_name = v1_lag.s_company_name and
v1.s_company_name = v1_lead.s_company_name and
v1.rn = v1_lag.rn + 1 and
v1.rn = v1_lead.rn - 1)
select * from v2
where d_year = 1999 and
avg_monthly_sales > 0 and
case when avg_monthly_sales > 0 then abs(sum_sales - avg_monthly_sales) / avg_monthly_sales else null end > 0.1
order by sum_sales - avg_monthly_sales, 3
limit 100
正如我们所见，该表v1在查询中被使用了 3 次
...
from v1, v1 v1_lag, v1 v1_lead
...
Web UI 中的图形如下
正如我们在左图中看到的那样，表的值number of output rows等于表store_sales的2,879,789大小。
但是，在右图中，它显示number of output rows同一张表的 等于5,759,578并且该值传播到下一个计划，例如Filter。
我们可以通过更简单的查询来获得相同的结果。
// create a temp table for tests
Seq(1, 2, 3).toDF("id").createOrReplaceTempView("t")
// execute the query
spark.sql("""
with v1 as (
select id
from t group by id)
select v1.id, v11.id id1, v12.id id2
from v1, v1 v11, v1 v12
where v1.id = v11.id and
v1.id = v12.id + 1
""").count
该查询的图表如下
正如我们所看到的，number of output rows它比表格的大小高两倍。此外，如果我们再添加一次表 v1 number of output rows，则表的大小是表的三倍，以此类推。
例如，如果我们像这样更改查询
...
select v1.id, v11.id id1, v12.id id2, v13.id id3
from v1, v1 v11, v1 v12, v1 v13
where v1.id = v11.id and
v1.id = v12.id + 1 and
v1.id = v13.id
...
number of output rows变成 9 。
值得一提的是，如果我们v1只使用该表两次，则number of output rows变为等于表大小。
所以，像这样的查询
...
select v1.id, v11.id id1
from v1, v1 v11
where v1.id = v11.id
...
number of output rows变成 3 。
在这样的情况下，我希望 Spark 加载表的次数与需要的表一样多次，或者加载表一次，然后在需要时重用它，但似乎我的两个假设都是错误的。
那么，为什么输出行数高于表大小？
我已经在 Spark 2.2 和 2.3 中对此进行了测试。