为什么在使用HBase时，persist(StorageLevel.MEMORY_AND_DISK)和cache()会产生不同的结果？

Question

为什么在使用HBase时，persist(StorageLevel.MEMORY_AND_DISK)和cache()会产生不同的结果？

apache-sparkjava-8apache-spark-sqlhbaseapache-spark-1.6

3

我可能听起来很幼稚，但这是我最近在项目中遇到的问题。需要更好地理解它。

df.persist(StorageLevel.MEMORY_AND_DISK)

每当我们在HBase读取时使用这样的persist - 对于流作业的其他后续批次，相同的数据会再次返回，但是HBase会在每个批次运行时进行更新。

HBase读取代码：

val df = sqlContext.read.options(Map(HBaseTableCatalog.tableCatalog -> schema)).format(dbSetup.dbClass).load().persist(StorageLevel.MEMORY_AND_DISK)

我用cache()代替了persist(StorageLevel.MEMORY_AND_DISK)，从HBase表返回的更新记录与预期一样。

我们尝试使用persist(StorageLevel.MEMORY_AND_DISK)的原因是为了确保内存存储不会满，并且我们不会在执行特定流程时重新进行所有转换。 Spark版本 - 1.6.3 HBase版本 - 1.1.2.2.6.4.42-1

能有人解释一下这个，并帮助我更好地理解吗？

- Dasarathy D R

@JacekLaskowski - 根据要求！同时，我所说的其他后续批次是指，在 Spark 流处理作业的一组间隔之后的下一批次。 - Dasarathy D R

“HBase Read Code” 在 Spark Streaming 作业中如何使用？请在您的问题中包含代码。这是 foreach 还是类似的东西？有什么理由坚持使用 1.6.3 吗？我怀疑它是否会得到很多关注（如果有的话）。 - Jacek Laskowski

这是我也遇到的问题：https://stackoverflow.com/questions/51791008/spark-application-returns-different-results-based-on-different-executor-memory - Avishek Bhattacharya

@JacekLaskowski 我在我的Spark Streaming作业中也使用了相同的代码。没有foreach或类似的东西。为了提供一个我们如何读取HBase的示例，请参考下面的链接。 [link]https://github.com/hortonworks-spark/shc/blob/master/examples/src/main/scala/org/apache/spark/sql/execution/datasources/hbase/HBaseSource.scala请查看上面链接中的withCatalog方法。那应该会给你一个更好的想法。 - Dasarathy D R

@JacekLaskowski - 没有特别的理由坚持使用1.6.3版本 - 只是我们的集群是用那个特定版本构建的。此外，我希望我的回答或响应与您的问题一致。如果不是，请随时详细说明。 - Dasarathy D R

@AvishekBhattacharya，我已经查看了您分享的链接。非常抱歉这么晚才回复。由于我被调到另一个模块，所以有些延迟。现在我已经重新开始处理这项工作了。虽然我理解链接中提到的内容，但这是否是导致此行为的确切原因？或者我们还需要查找其他方面的问题。 - Dasarathy D R

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Aramis NSR · Accepted Answer

正如你提到的，你正在寻找一个"为什么"的理由，因此我回答这个问题，否则这个问题将无法得到解答，因为现在运行Spark 1.6.3来感知那个特定HBASE版本没有任何合理的理由。

在内部，当你使用cache()时，Spark会调用persist()，并且在RDDs和Datasets（或Dataframes）上的行为是不同的。在RDDs上，它使用MEMORY_ONLY，在Datasets上，它使用MEMORY_AND_DISK。我无法看到你（完整）编写的代码，但通常我可以说，你不应该遇到两种缓存和持久化方式之间的差异，而且你的问题只是版本不兼容或者可能是Apache没有修复的一个bug。

有几个地方可以检查出错的原因。

在这个链接https://spark.apache.org/releases/spark-release-1-6-3.html中，你可以发现代码的维护是在1.6分支上进行的，所以这里是找到代码的地方https://github.com/apache/spark/blob/branch-1.6/core/src/main/scala/org/apache/spark/CacheManager.scala。

希望对你有所帮助。