如何在 Spark 中显式实现 RDD答案

【问题标题】：How to materialize an RDD explicitly in Spark如何在 Spark 中显式实现 RDD
【发布时间】：2020-05-24 09:28:16
【问题描述】：

我需要从文件中读取一个大型数据集，将其转换为 Spark 矩阵并在该矩阵上运行一些机器学习算法。我想对机器学习算法的速度进行基准测试。由于 Spark RDD 总是被延迟评估，因此很难对机器学习算法进行基准测试。当我测量运行时，它还包括解析输入文件的运行时。

有没有办法强制 Spark 实现一些 RDD？这样我就可以在运行机器学习算法之前提前解析输入文件？

谢谢，大

【问题讨论】：

【解决方案1】：

我通常会这样做：

val persisted = rdd.persist(...);

这取决于你的 rdd 的大小，如果它适合内存只提供内存，否则 - 内存和磁盘级别。

然后：

persisted.count();
// now you can use 'persisted', it's materialized

然后是所有其他管道转换（在您的情况下为 ml）

所以 count 是一个动作 - 所以它实现了 rdd 并且因为你之前已经持久化了它 - 下一个阶段将从持久存储而不是从文件中获取 rdd

【讨论】：

如果你只在内存中持久化，你可以使用 val persisted = rdd.cache()。它与仅在内存中持久化具有完全相同的效果
我想这就是你的意思。我使用存储在内存中的数据设置 RDD 持久化并运行 count()。它确实会触发 Spark 解析输入文件。但是，当我对持久数据运行 corr() 时，我看不到任何加速。我在做正确的事吗？ lines = sc.textFile(sys.argv[1]) data = lines.map(parseVector) data1 = data.persist(storageLevel=StorageLevel.MEMORY_ONLY) data1.count() start_time = time.time() corr = Statistics.corr (data1, method=corrType) end_time = time.time() print("%s seconds" % (end_time - start_time))
您是否验证过您的文件可以存储在内存中？可能瓶颈在于 ml 而不是解析。
我想我需要增加 JVM 的堆大小（现在堆大小比数据集大小大得多）。在我这样做之后，我确实看到了一些性能改进。

【解决方案2】：

如果您的应用程序 (ML) 有多个使用 data1.count 的任务（参考 @Igor 的 cmets 部分），您也可以使用共享变量和累加器参考这个source

【讨论】：