查询计划

Python 中的 Spark SQL 入门

Mark Plutowski

Data Scientist

Explain（解释）

EXPLAIN SELECT * FROM table1

加载数据框并注册

df = spark.read.load('/temp/df.parquet')

df.registerTempTable('df')

运行 EXPLAIN 查询

spark.sql('EXPLAIN SELECT * FROM df').first()

Row(plan='== Physical Plan ==\n
*FileScan parquet [word#1928,id#1929L,title#1930,part#1931] 
  Batched: true, 
  Format: Parquet, 
  Location: InMemoryFileIndex[file:/temp/df.parquet], 
  PartitionFilters: [], 
  PushedFilters: [], 
  ReadSchema: struct<word:string,id:bigint,title:string,part:int>')

解读 EXPLAIN 查询

== 物理计划 ==

FileScan parquet [word#1928,id#1929L,title#1930,part#1931]
Batched: true,
Format: Parquet,
Location: InMemoryFileIndex[file:/temp/df.parquet],
PartitionFilters: [],
PushedFilters: [],
ReadSchema: struct<word:string,id:bigint,title:string,part:int>'

df.explain()

df.explain()

== 物理计划 ==
FileScan parquet [word#963,id#964L,title#965,part#966] 
Batched: true, Format: Parquet, 
Location: InMemoryFileIndex[file:/temp/df.parquet], 
PartitionFilters: [], PushedFilters: [], 
ReadSchema: struct<word:string,id:bigint,title:string,part:int>

spark.sql("SELECT * FROM df").explain()

== 物理计划 ==
FileScan parquet [word#712,id#713L,title#714,part#715] 
Batched: true, Format: Parquet, 
Location: InMemoryFileIndex[file:/temp/df.parquet], 
PartitionFilters: [], PushedFilters: [], 
ReadSchema: struct<word:string,id:bigint,title:string,part:int>

缓存数据框上的 df.explain()

df.cache()
df.explain()

== 物理计划 ==
InMemoryTableScan [word#0, id#1L, title#2, part#3]
   +- InMemoryRelation [word#0, id#1L, title#2, part#3], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas)
      +- FileScan parquet [word#0,id#1L,title#2,part#3] 
         Batched: true, Format: Parquet, Location: 
         InMemoryFileIndex[file:/temp/df.parquet], 
         PartitionFilters: [], PushedFilters: [], 
         ReadSchema: struct<word:string,id:bigint,title:string,part:int>

spark.sql("SELECT * FROM df").explain()

== 物理计划 ==
InMemoryTableScan [word#0, id#1L, title#2, part#3]
   +- InMemoryRelation [word#0, id#1L, title#2, part#3], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas)
     +- FileScan parquet [word#0,id#1L,title#2,part#3] 
        Batched: true, Format: Parquet, 
        Location: InMemoryFileIndex[file:/temp/df.parquet], 
        PartitionFilters: [], PushedFilters: [], 
        ReadSchema: struct<word:string,id:bigint,title:string,part:int>

按词频排序的查询

SELECT word, COUNT(*) AS count 
FROM df 
GROUP BY word 
ORDER BY count DESC

等效的点式写法：

df.groupBy('word')
  .count()
  .sort(desc('count'))
  .explain()

使用数据框点式写法的同一查询

== 物理计划 ==
*Sort [count#1040L DESC NULLS LAST], true, 0
+- Exchange rangepartitioning(count#1040L DESC NULLS LAST, 200)
   +- *HashAggregate(keys=[word#963], functions=[count(1)])
      +- Exchange hashpartitioning(word#963, 200)
         +- *HashAggregate(keys=[word#963], functions=[partial_count(1)])
            +- InMemoryTableScan [word#963]
                  +- InMemoryRelation [word#963, id#964L, title#965, part#966],
                     true,10000, StorageLevel(disk, memory, deserialized,
                     1 replicas)
                        +- *FileScan parquet [word#963,id#964L,title#965,part#966]
                           Batched: true, Format: Parquet,
                           Location: InMemoryFileIndex[file:/temp/df.parquet],
                           PartitionFilters: [], PushedFilters: [],
                           ReadSchema: struct<word:string,id:bigint,title:string,part:int>

自下而上阅读

FileScan parquet
InMemoryRelation
InMemoryTableScan
HashAggregate(keys=[word#963], ...)`
HashAggregate(keys=[word#963], functions=[count(1)])`
Sort [count#1040L DESC NULLS LAST]`

查询计划

== 物理计划 ==
*Sort [count#1160L DESC NULLS LAST], true, 0
+- Exchange rangepartitioning(count#1160L DESC NULLS LAST, 200)
   +- *HashAggregate(keys=[word#963], functions=[count(1)])
      +- Exchange hashpartitioning(word#963, 200)
         +- *HashAggregate(keys=[word#963], functions=[partial_count(1)])
            +- *FileScan parquet [word#963] Batched: true, Format: Parquet,
               Location: InMemoryFileIndex[file:/temp/df.parquet], PartitionFilters: [],
               PushedFilters: [], ReadSchema: struct<word:string>

先前的计划包含以下行，而上面的计划中缺失：

...
            +- InMemoryTableScan [word#963]
                  +- InMemoryRelation [word#963, id#964L, title#965, part#966], true, 10000,
                     StorageLevel(disk, memory, deserialized, 1 replicas)
...

Passons à la pratique !

Python 中的 Spark SQL 入门