我有两个大的配置单元表，我想用spark.sql连接它们。假设我们有表1和表2，表1有500万行，表2有7000万行。表是snappy格式的，并作为Parquet文件存储在配置单元中。
我想加入它们，并对某些列进行一些聚合，比如说计数所有行和列的平均值（例如doublecolumn），同时使用两个条件（比如col1、col2）进行过滤。
注意：我在一台机器上进行测试安装（虽然功能非常强大）。我预计集群中的性能可能会有所不同。
我的第一次尝试是使用spark sql，例如：

val stat = sqlContext.sql("select count(id), avg(doubleColumn) " +
                              " FROM db.table1 as t1 JOIN db.table2 " +
                              " ON t1.id = t2.id " + 
                              " WHERE col1 = val1 AND col2 = val2").collect

不幸的是，即使我给每个执行器和驱动程序至少8gb内存，这个5分钟的运行也非常糟糕。我还尝试使用dataframe语法，并尝试先过滤行，然后只选择特定的列以获得更好的选择性，如：

//Filter first and select only needed column
val df = spark.sql("SELECT * FROM db.tab1")
val tab1= df.filter($"col1" === "val1" && $"col2" === "val2").select("id")

val tab2= spark.sql("SELECT id, doubleColumn FROM db.tab2")
val joined = tab1.as("d1").join(tab2.as("d2"), $"d1.id" === $"d2.id") 

//Take the aggregations on the joined df
import org.apache.spark.sql.functions;

joined.agg(
   functions.count("id").as("count"),
   functions.avg("doubleColumn").as("average")
).show();

但这并没有显著的性能提升。如何提高join的性能？
执行spark.sql或dataframe语法的最佳方法是什么？
给更多的执行者或记忆会有帮助吗？
我应该使用缓存吗？
我缓存了dataframes tab1、tab2和join aggregation，它们都有显著的收益，但我认为缓存我的dataframes并不实用，因为我们对并发感兴趣，许多用户同时询问一些分析性查询。
是不是因为我在单节点上工作，而当我在集群上的生产环境中工作时，我的问题就会消失？
额外的问题：我用impala尝试了这个查询，它做了大约40秒，但是它比spark.sql好得多。 Impala 怎么能比星火更好？！