当 rdd 项目很大时，为什么 rdd.map(identity).cache 很慢？

【问题标题】：Why is rdd.map(identity).cache slow when rdd items are big?当 rdd 项目很大时，为什么 rdd.map(identity).cache 很慢？
【发布时间】：2016-10-17 23:34:36
【问题描述】：

我发现在 rdd 上使用 .map( identity ).cache 时，如果项目很大，它会变得非常慢。虽然它几乎是瞬时的。

注意：这可能与this question有关，但这里我提供一个非常精确的例子（可以直接在spark-shell中执行）：

// simple function to profile execution time (in ms)
def profile[R](code: => R): R = {
  val t = System.nanoTime
  val out = code
  println(s"time = ${(System.nanoTime - t)/1000000}ms")
  out
}

// create some big size item
def bigContent() = (1 to 1000).map( i => (1 to 1000).map( j => (i,j) ).toMap )

// create rdd
val n = 1000 // size of the rdd

val rdd = sc.parallelize(1 to n).map( k => bigContent() ).cache
rdd.count // to trigger caching

// profiling
profile( rdd.count )                 // around 12 ms
profile( rdd.map(identity).count )   // same
profile( rdd.cache.count )           // same
profile( rdd.map(identity).cache.count ) // 5700 ms !!!

我首先预计是时候创建一个新的 rdd（容器）了。但是如果我使用大小相同但内容很少的rdd，则执行时间只有微小的差异：

val rdd = parallelize(1 to n).cache
rdd.count

profile( rdd.count )                 // around 9 ms
profile( rdd.map(identity).count )   // same
profile( rdd.cache.count )           // same
profile( rdd.map(identity).cache.count ) // 15 ms

所以，看起来缓存实际上是在复制数据。我认为它也可能会浪费时间序列化它，但我检查了缓存是否与默认的 MEMORY_ONLY 持久性一起使用：

rdd.getStorageLevel == StorageLevel.MEMORY_ONLY // true

=> 那么，是缓存复制数据，还是别的什么？

这确实是我的应用程序的一个主要限制，因为我从一个使用类似于rdd = rdd.map(f: Item => Item).cache 的设计开始，它可以与许多这样的函数一起使用 f 以任意顺序应用（我无法事先确定的顺序）。

我使用的是 Spark 1.6.0

编辑

当我查看 spark ui -> 阶段选项卡 -> 最后一个阶段（即 4）时，所有任务都有几乎相同的数据：

  • duration = 3s（下降到 3s，但这仍然是 2.9 太多了：-\）
  • 调度器 10ms
  • 任务反序列化 20ms
  • gc 0.1s（所有任务都有，但是为什么会触发gc？？？）
  • 结果序列化 0ms
  • 0ms 得到结果
  • 峰值执行内存 0.0B
  • 输入大小 7.0MB/125
  • 没有错误

【问题讨论】：

  • 缓存意味着 spark 会将你的数据复制到列数据中的 spark 缓存中。并且您的操作是缓存之前的映射，可能会导致随机播放操作。您可以通过 Spark UI 检查您的地图和缓存时发生了什么。
  • 柱状数据是什么意思？对于一些A，这里只是RDD[A]。此外，没有洗牌。这是最简单的地图操作。如果地图很复杂，则在第二个配置文件中需要时间：rdd.map(identity).count
  • 我会查看 UI，但我不确定要查找什么
  • 这意味着 Spark 将像 parquet 文件一样以列格式存储数据（列格式）。在 Spark UI 中，转到 Stages 选项卡（域/阶段）-> 参见表 Summary Metrics for 35 Completed Tasks
  • @Juh_ 这很正常，因为您要求缓存一个新的 rdd（已由 .map(identity) 转换。因此，如果您附加缓存然后对其计数，将触发第一个计数缓存机制，但您不会在此特定计数操作中从中受益。但是，对缓存的 rdd 执行以下操作将利用它作为缓存 - 前提是您将对它的引用存储在变量中。

标签： performance caching apache-spark


【解决方案1】：

在慢速缓存期间运行org.apache.spark.executor.CoarseGrainedExecutorBackend 的进程的jstack 显示以下内容：

"Executor task launch worker-4" #76 daemon prio=5 os_prio=0 tid=0x00000000030a4800 nid=0xdfb runnable [0x00007fa5f28dd000]
   java.lang.Thread.State: RUNNABLE
  at java.util.IdentityHashMap.resize(IdentityHashMap.java:481)
  at java.util.IdentityHashMap.put(IdentityHashMap.java:440)
  at org.apache.spark.util.SizeEstimator$SearchState.enqueue(SizeEstimator.scala:176)
  at org.apache.spark.util.SizeEstimator$.visitArray(SizeEstimator.scala:251)
  at org.apache.spark.util.SizeEstimator$.visitSingleObject(SizeEstimator.scala:211)
  at org.apache.spark.util.SizeEstimator$.org$apache$spark$util$SizeEstimator$$estimate(SizeEstimator.scala:203)
  at org.apache.spark.util.SizeEstimator$$anonfun$sampleArray$1.apply$mcVI$sp(SizeEstimator.scala:284)
  at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:141)
  at org.apache.spark.util.SizeEstimator$.sampleArray(SizeEstimator.scala:276)
  at org.apache.spark.util.SizeEstimator$.visitArray(SizeEstimator.scala:260)
  at org.apache.spark.util.SizeEstimator$.visitSingleObject(SizeEstimator.scala:211)
  at org.apache.spark.util.SizeEstimator$.org$apache$spark$util$SizeEstimator$$estimate(SizeEstimator.scala:203)
  at org.apache.spark.util.SizeEstimator$.estimate(SizeEstimator.scala:70)
  at org.apache.spark.util.collection.SizeTracker$class.takeSample(SizeTracker.scala:78)
  at org.apache.spark.util.collection.SizeTracker$class.afterUpdate(SizeTracker.scala:70)
  at org.apache.spark.util.collection.SizeTrackingVector.$plus$eq(SizeTrackingVector.scala:31)
  at org.apache.spark.storage.MemoryStore.unrollSafely(MemoryStore.scala:285)
  at org.apache.spark.CacheManager.putInBlockManager(CacheManager.scala:171)
  at org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:78)
  at org.apache.spark.rdd.RDD.iterator(RDD.scala:268)
  at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
  at org.apache.spark.scheduler.Task.run(Task.scala:89)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  at java.lang.Thread.run(Thread.java:745)


"Executor task launch worker-5" #77 daemon prio=5 os_prio=0 tid=0x00007fa6218a9800 nid=0xdfc runnable [0x00007fa5f34e7000]
   java.lang.Thread.State: RUNNABLE
  at java.util.IdentityHashMap.put(IdentityHashMap.java:428)
  at org.apache.spark.util.SizeEstimator$SearchState.enqueue(SizeEstimator.scala:176)
  at org.apache.spark.util.SizeEstimator$$anonfun$visitSingleObject$1.apply(SizeEstimator.scala:224)
  at org.apache.spark.util.SizeEstimator$$anonfun$visitSingleObject$1.apply(SizeEstimator.scala:223)
  at scala.collection.immutable.List.foreach(List.scala:318)
  at org.apache.spark.util.SizeEstimator$.visitSingleObject(SizeEstimator.scala:223)
  at org.apache.spark.util.SizeEstimator$.org$apache$spark$util$SizeEstimator$$estimate(SizeEstimator.scala:203)
  at org.apache.spark.util.SizeEstimator$.estimate(SizeEstimator.scala:70)
  at org.apache.spark.util.collection.SizeTracker$class.takeSample(SizeTracker.scala:78)
  at org.apache.spark.util.collection.SizeTracker$class.afterUpdate(SizeTracker.scala:70)
  at org.apache.spark.util.collection.SizeTrackingVector.$plus$eq(SizeTrackingVector.scala:31)
  at org.apache.spark.storage.MemoryStore.unrollSafely(MemoryStore.scala:285)
  at org.apache.spark.CacheManager.putInBlockManager(CacheManager.scala:171)
  at org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:78)
  at org.apache.spark.rdd.RDD.iterator(RDD.scala:268)
  at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
  at org.apache.spark.scheduler.Task.run(Task.scala:89)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
  at java.lang.Thread.run(Thread.java:745)

SizeEstimator 作为缓存表面上已经在内存中的东西的主要成本之一是有意义的，因为对未知对象进行正确的大小估计可能相当困难；如果查看visitSingleObject 方法，您会发现它严重依赖反射，调用getClassInfo 访问运行时类型信息；不仅遍历了完整的对象层次结构，而且每个嵌套成员都会根据IdentityHashMap 进行检查，以检测哪些引用引用了相同的具体对象实例，因此堆栈跟踪显示了这些 IdentityHashMap 操作中的大量时间。

在您的示例对象的情况下，您基本上将每个项目作为从包装整数到包装整数的映射列表；大概 Scala 的内部映射实现也包含一个数组，这解释了 visitSingleObject -> List.foreach -> visitSingleObject -> visitSingleObject 调用层次结构。无论如何，在这种情况下，要访问 很多 个内部对象，并且 SizeEstimators 为每个采样的对象设置一个新的 IdentityHashMap。

在您测量的情况下：

profile( rdd.cache.count )

这不算在执行缓存逻辑，因为 RDD 已经成功缓存，所以 Spark 足够聪明，不会重新运行缓存逻辑。实际上，您可以通过直接分析新的 RDD 创建和缓存来独立于额外的“map(identity)”转换来隔离缓存逻辑的确切成本；这是我的 Spark 会话，从您的最后几行开始：

scala> profile( rdd.count )
time = 91ms
res1: Long = 1000

scala> profile( rdd.map(identity).count )
time = 112ms
res2: Long = 1000

scala> profile( rdd.cache.count )
time = 59ms
res3: Long = 1000

scala> profile( rdd.map(identity).cache.count )
time = 6564ms                                                                   
res4: Long = 1000

scala> profile( sc.parallelize(1 to n).map( k => bigContent() ).count )
time = 14990ms                                                                  
res5: Long = 1000

scala> profile( sc.parallelize(1 to n).map( k => bigContent() ).cache.count )
time = 22229ms                                                                  
res6: Long = 1000

scala> profile( sc.parallelize(1 to n).map( k => bigContent() ).map(identity).cache.count )
time = 21922ms                                                                  
res7: Long = 1000

所以你可以看到，缓慢并不是因为你运行了map 转换，而是在这种情况下，~6s 似乎是计算 1000 缓存逻辑的基本成本当每个对象具有大约 1,000,000 到 10,000,000 个内部对象时的对象（取决于 Map 实现的布局方式；例如，顶部堆栈跟踪中的额外 visitArray 嵌套暗示 HashMap impl 具有嵌套数组，这对于每个哈希表条目内的典型密集线性探测数据结构）。

对于您的具体用例，如果可能，您应该在延迟缓存方面犯错，因为缓存中间结果会产生开销，如果您真的不打算将中间结果用于大量单独的下游转换。但是正如您在问题中提到的那样，如果您确实使用一个 RDD 来分支到多个不同的下游转换，那么如果原始转换非常昂贵，您可能确实需要缓存步骤。

解决方法是尝试使用更适合恒定时间计算的内部数据结构（例如基元数组），您可以在其中节省大量成本以避免迭代大量数字包装器对象的数量，并取决于它们在 SizeEstimator 中的反射。

我尝试了 Array[Array[Int]] 之类的方法，尽管开销仍然不为零，但对于类似的数据大小，它的性能要好 10 倍：

scala> def bigContent2() = (1 to 1000).map( i => (1 to 1000).toArray ).toArray
bigContent2: ()Array[Array[Int]]

scala> val rdd = sc.parallelize(1 to n).map( k => bigContent2() ).cache
rdd: org.apache.spark.rdd.RDD[Array[Array[Int]]] = MapPartitionsRDD[23] at map at <console>:28

scala> rdd.count // to trigger caching
res16: Long = 1000                                                              

scala> 

scala> // profiling

scala> profile( rdd.count )
time = 29ms
res17: Long = 1000

scala> profile( rdd.map(identity).count )
time = 42ms
res18: Long = 1000

scala> profile( rdd.cache.count )
time = 34ms
res19: Long = 1000

scala> profile( rdd.map(identity).cache.count )
time = 763ms                                                                    
res20: Long = 1000

为了说明反射任何更高级对象的成本有多糟糕，如果我删除最后一个 toArray 并最终使每个 bigContent 成为 scala.collection.immutable.IndexedSeq[Array[Int]]，则性能回到 ~2x 以内原IndexSeq[Map[Int,Int]]案例的缓慢：

scala> def bigContent3() = (1 to 1000).map( i => (1 to 1000).toArray )
bigContent3: ()scala.collection.immutable.IndexedSeq[Array[Int]]

scala> val rdd = sc.parallelize(1 to n).map( k => bigContent3() ).cache
rdd: org.apache.spark.rdd.RDD[scala.collection.immutable.IndexedSeq[Array[Int]]] = MapPartitionsRDD[27] at map at <console>:28

scala> rdd.count // to trigger caching
res21: Long = 1000                                                              

scala> 

scala> // profiling

scala> profile( rdd.count )
time = 27ms
res22: Long = 1000

scala> profile( rdd.map(identity).count )
time = 39ms
res23: Long = 1000

scala> profile( rdd.cache.count )
time = 37ms
res24: Long = 1000

scala> profile( rdd.map(identity).cache.count )
time = 2781ms                                                                   
res25: Long = 1000

正如评论部分所讨论的，您还可以考虑使用 MEMORY_ONLY_SER StorageLevel，只要有一个高效的序列化器，它很可能比 SizeEstimator 中使用的递归反射便宜；为此，您只需将cache() 替换为persist(StorageLevel.MEMORY_ONLY_SER)；正如this other question 中提到的，cache() 在概念上与persist(StorageLevel.MEMORY_ONLY) 相同。

import org.apache.spark.storage.StorageLevel
profile( rdd.map(identity).persist(StorageLevel.MEMORY_ONLY_SER).count )

我实际上在 Spark 1.6.1 和 Spark 2.0.0-preview 上都试过这个，运行时集群配置的其他一切都完全相同（分别使用Google Cloud Dataproc 的“1.0”和“预览”图像版本）。不幸的是，MEMORY_ONLY_SER 技巧在 Spark 1.6.1 中似乎没有帮助：

scala> profile( rdd.map(identity).persist(StorageLevel.MEMORY_ONLY_SER).count )
time = 6709ms                                                                   
res19: Long = 1000

scala> profile( rdd.map(identity).cache.count )
time = 6126ms                                                                   
res20: Long = 1000

scala> profile( rdd.map(identity).persist(StorageLevel.MEMORY_ONLY).count )
time = 6214ms                                                                   
res21: Long = 1000

但在 Spark 2.0.0-preview 中，它似乎将性能提高了 10 倍：

scala> profile( rdd.map(identity).persist(StorageLevel.MEMORY_ONLY_SER).count )
time = 500ms
res18: Long = 1000

scala> profile( rdd.map(identity).cache.count )
time = 5353ms                                                                   
res19: Long = 1000

scala> profile( rdd.map(identity).persist(StorageLevel.MEMORY_ONLY).count )
time = 5927ms                                                                   
res20: Long = 1000

不过，这可能会因您的对象而异；只有在序列化本身不使用大量反射的情况下，才能预期加速；如果您能够有效地使用Kryo serialization，那么您很可能会看到使用MEMORY_ONLY_SER 对这些大型对象的改进。

【讨论】：

  • 但结论是骗人的：-\ 我的目标是运行许多rdd = rdd.map(f).cache，然后收集一些统计信息（例如rdd.map( _.map(_.sum) ).collect）。然后任意选择以下f，部分w.r.t 这些收集的值。有些 f 很简单，有些很长......不过，您展示了使用数组提高效率的好处。但是不知道能不能用。我的真实结构更复杂。我需要做一些测试。
  • 我还需要看看Spark的SizeEstimator系统。乍一看，我似乎可以制作一个实现它的自制类来绕过完全解析。有这方面的知识吗？
  • 不确定是否可以动态执行，虽然我之前确实构建了 Spark 程序集的自定义，但如果您只想 git clone 并构建自己的 Spark，说明非常好：@ 987654326@.
  • 但是在深入研究 Spark 层之前，您还可以考虑将存储级别更改为 MEMORY_ONLY_SER 而不是 MEMORY_ONLY；我还没有时间测试它，但理论上以序列化形式存储缓存层至少会绕过重量级大小估计逻辑，实际上就像你手动将所有内容保存为字节数组格式以在内部重组一样任务逻辑。而且根据我过去的经验，反序列化成本通常比进行过多的反射要便宜。
  • 有趣的是，我在 Spark 1.6.1 和 Spark 2.0.0-preview 中使用Google Cloud Dataproc 尝试了 MEMORY_ONLY_SER，而 Spark 1.6.1 显示 MEMORY_ONLY_SER 与缓存或持久化（MEMORY_ONLY）的性能相同，而在 Spark 2.0.0-preview 中，MEMORY_ONLY_SER 快了 10 倍。
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 2019-10-02
  • 2010-10-30
  • 2023-04-03
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2011-05-24
相关资源
最近更新 更多
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode