performance - 连接的 Spark 性能分析

标签 performance apache-spark bigdata distributed-computing apache-spark-sql

输入数据

我有两个从 MySQL 导出为 csv 文件的表。

磁盘上的表 1 大小:250 MB
记录:70 万

磁盘上的表 2 大小:350 MB
记录:60 万

更新代码

import org.apache.spark.sql.SQLContext
val sqlContext = new SQLContext(sc)
val table-one = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").load("example-input-files/table-1-data.csv”)
table-one.registerTempTable(“table-one”)
val table-two = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").load("example-input-files/table-2-data.csv”)
table-two.registerTempTable(“table”-two)
sqlContext.cacheTable(“table-one”)
sqlContext.cacheTable(“table-two”)
val result = sqlContext.sql("SELECT table-one.ID,table-two.ID FROM table-one LEFT JOIN table-two ON table-one.ID = table-two.ID")
result.take(2).foreach(println)

Spark 工作
  • 使用 Databricks CSV lib 读取两个 csv 文件并将它们注册为
    表。
  • 使用公共(public)列对两者执行左连接,典型的左
    加入关系数据库发言。
  • 打印前两个结果,因为在控制台本身打印会
    消耗时间。

  • 这总共需要 30 秒。我在一台具有足够内存的机器上运行,以便两个文件都可以放入(毕竟它是 600Mb)。

    我有两种方式来完成这项工作。
  • 整体运行作业,即加载所有 csv,运行连接,然后打印结果
  • 第二种方法是我第一次使用 sqlContext.cacheTable("the_table") 在内存中运行和缓存表。

  • 缓存后我发现加入操作本身需要 8 秒才能完成。

    这个时间合理吗?我猜它不是,可以做很多优化来加快查询速度。

    我看到的优化
  • 将数据放入 HDFS 而不是本地磁盘。这会加快检索速度吗?
  • 在集群上运行,我猜这不会很快,因为数据可以放入内存并且顺序会更快。
  • 数据建模和使用 cassandra 会更快吗?
  • 我正在使用纯 SQL 加入,RDD 加入会更快吗?

  • 还有其他方法可以做得更好吗?

    最佳答案

    正如评论者所提到的,Spark 专为分布式计算而设计。与其他 PL 相比,在本地处理小型(ish)数据时,仅所有初始化和调度的开销就足以使 Spark 看起来很慢。

    Running on a cluster,I am guessing that this will not be fast since the data can fit into memory and sequential will be faster.



    只要您的代码执行狭窄的转换,执行程序实际上就会在内存中的本地数据副本上工作,因此这并不完全正确。但是,您的代码执行连接,这是一个广泛的转换 - 这意味着必须在网络上对 block 进行洗牌。请记住这一点。广泛的转换是昂贵的,因此尽可能将它们放在 DAG 的末尾。但同样,您的数据足够小,您可能看不到好处。

    另一件事是,如果您有 Hive,那么您可以考虑将数据存储在连接列上分区的表中。

    关于performance - 连接的 Spark 性能分析,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/31849599/

    相关文章:

    hadoop - 如何获取基于 yarn 的 Spark 应用程序的 CPU 使用情况

    performance - javaFX 内存释放,javaFX 错误?

    .net - System.Activator.CreateInstance(T) 是否存在足以阻止我们随意使用它的性能问题?

    python - 使用 python 进行 Spark 流处理时出现错误?

    python - Spark ML Pipeline 导致 java.lang.Exception : failed to compile . .. 代码 ... 超出 64 KB

    scala - Spark/Scala 中 array.map 和 rdd.map 有什么区别?

    r - 如何在Python中使用R studio中已有的贝叶斯变化点库作为bcp?

    c++ - 我可以在类中声明一个非常大的数组吗,C++

    SQL:链接连接效率

    sql-server - 有效更新 980K 记录,其中行在组中具有最大值