Spark SQL повторяет вычисление того же подзапроса при объединении - PullRequest
Новая
       6

Spark SQL повторяет вычисление того же подзапроса при объединении

0 голосов
/ 23 февраля 2019

У меня есть объединение двух абсолютно одинаковых подзапросов.Однако, исходя из объяснения запроса, Spark SQL, кажется, выполняет один и тот же подзапрос дважды.Ожидаемый? 

    In [20]: session.sql('(select count(city_code) as c from location group by country_code having c < 10) union (select count(city_code) as c from location group by country_code having c < 10)').explain(True)


== Parsed Logical Plan ==
'Distinct
+- 'Union
   :- 'Filter ('c < 10)
   :  +- 'Aggregate ['country_code], ['count('city_code) AS c#228]
   :     +- 'UnresolvedRelation `location`
   +- 'Filter ('c < 10)
      +- 'Aggregate ['country_code], ['count('city_code) AS c#229]
         +- 'UnresolvedRelation `location`

== Analyzed Logical Plan ==
c: bigint
Distinct
+- Union
   :- Filter (c#228L < cast(10 as bigint))
   :  +- Aggregate [country_code#231], [count(city_code#234) AS c#228L]
   :     +- SubqueryAlias location
   :        +- Relation[uid#230L,country_code#231,country_category#232,region_code#233,city_code#234,time#235L] parquet
   +- Filter (c#229L < cast(10 as bigint))
      +- Aggregate [country_code#237], [count(city_code#240) AS c#229L]
         +- SubqueryAlias location
            +- Relation[country_code#237,city_code#240] parquet

== Optimized Logical Plan ==
Aggregate [c#228L], [c#228L]
+- Union
   :- Filter (c#228L < 10)
   :  +- Aggregate [country_code#231], [count(city_code#234) AS c#228L]
   :     +- Project [country_code#231, city_code#234]
   :        +- Relation[country_code#231,city_code#234] parquet
   +- Filter (c#229L < 10)
      +- Aggregate [country_code#237], [count(city_code#240) AS c#229L]
         +- Project [country_code#237, city_code#240]
            +- Relation[country_code#237,city_code#240] parquet

== Physical Plan ==
*HashAggregate(keys=[c#228L], functions=[], output=[c#228L])
+- Exchange hashpartitioning(c#228L, 200)
   +- *HashAggregate(keys=[c#228L], functions=[], output=[c#228L])
      +- Union
         :- *Filter (c#228L < 10)
         :  +- *HashAggregate(keys=[country_code#231], functions=[count(city_code#234)], output=[c#228L])
         :     +- Exchange hashpartitioning(country_code#231, 200)
         :        +- *HashAggregate(keys=[country_code#231], functions=[partial_count(city_code#234)], output=[country_code#231, count#255L])
         :           +- *FileScan parquet default.location[country_code#231,city_code#234] Batched: true, Format: Parquet, Location: InMemoryFileIndex[.../location], PartitionFilters: [], PushedFilters: [],
 ReadSchema: struct<country_code:string,city_code:string>
         +- *Filter (c#229L < 10)
            +- *HashAggregate(keys=[country_code#237], functions=[count(city_code#240)], output=[c#229L])
               +- ReusedExchange [country_code#237, count#257L], Exchange hashpartitioning(country_code#231, 200)

Ответы [ 2 ]

0 голосов
/ 23 февраля 2019 
session.sql('(select count(city_code) as c from location 
              group by country_code having c < 10) 
             union 
             (select count(city_code) as c from location 
              group by country_code having c < 10)')

В этом подходе - как показывает pp - он не учитывает кэширование.Вам нужен какой-то внешний аспект, чтобы сослаться на него, и вуаля мне предшествует более ранний ответ.

Хотя ваш вопрос: таким образом, да, это ожидается.Обратитесь к другому ответу, а также тому, что я также заявляю, и о его экземплярах.

Примечание: предложение WITH также не может быть кэшировано.

0 голосов
/ 23 февраля 2019

Краткий ответ: да.

Рассмотрим пример кода scala spark 

val data = sqlContext.read.table("location")
val a = data.groupBy("country_code").agg(count(city_code) as "c").filter($"c" < 10)
val b = data.groupBy("country_code").agg(count(city_code) as "c").filter($"c" < 10)
a.union(b).show()

Вы все еще удивлены ?.

Таким образом, будет легче увидеть, что у spark будут два дерева выполнения, одно для val a и одно для val b.

Оптимизация Spark не заботится о перекрытиях между двумя запросами, даже если естьпрактически одинаковы.Он будет вычислять поддерево для каждого запроса и оптимизировать это поддерево.

...