python - Pyspark HiveContext.table 和 HiveContext.sql 性能

Question

我想知道spark如何处理这些情况以及使用什么会更快更好。数据帧中的原生计数和 Spark sql 中的计数哪个更好？我认为这不仅仅涉及计数。

hc = HiveContext(sc)

t0 = time()
a = hc.table('default.testtable')
c = a.count()
tt = time() - t0
print(c)
print("Count for 1st case completed in {} seconds".format(round(tt,3)))

t0 = time()
b = hc.sql('select count(*) from default.testtable')
c1 = b.collect()
tt = time() - t0
print(c1)
print("Count for 2nd case completed in {} seconds".format(round(tt,3)))

如果我在单次运行中执行它，它会给出:

77707920
Count for 1st case completed in 48.358 seconds
[Row(_c0=77707920)]
Count for 2nd case completed in 20.895 seconds

如果我单独执行，在评论第二个然后是第一个 block 之前，然后是第一个更快:

77707920
Count for 1st case completed in 18.717 seconds

[Row(_c0=77707920)]
Count for 2nd case completed in 19.074 seconds

两者的物理计划相似:

对于收集:

== Physical Plan ==
TungstenAggregate(key=[], functions=[(count(1),mode=Final,isDistinct=false)], output=[_c0#9L])
+- TungstenExchange SinglePartition, None
   +- TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=[count#12L])
      +- HiveTableScan MetastoreRelation default, testtable, None

计数:

== Physical Plan ==
TungstenAggregate(key=[], functions=[(count(1),mode=Final,isDistinct=false)], output=[count#9L])
+- TungstenExchange SinglePartition, None
   +- TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=[count#12L])
      +- HiveTableScan MetastoreRelation default, testtable, None

Spark内部是如何对待它的？为什么性能结果不同？使用哪个 native 数据帧计数或 Spark sql 计数更好？

Answer 1

没有什么区别。您在第一种情况下的测量结果是错误的。

• 第一个时间 block 包括上下文初始化时间、执行器分配时间和一堆其他辅助任务。
• 当一切准备就绪时，第二个时间段就会到来。

如果颠倒顺序，您应该会看到时间相反，sql 比 table 更快。​​

由于某些 JVM 和系统优化，您还可能会看到重复执行带来的一些性能改进。

如果您想获得可靠的估计，请分别为每个场景运行程序，并使用 timeit 多次重复计时。