abhishekd0907 commented on a change in pull request #29242:
URL: https://github.com/apache/spark/pull/29242#discussion_r464232887
##########
File path: python/pyspark/sql/dataframe.py
##########
@@ -674,7 +674,7 @@ def cache(self):
.. note:: The default storage level has changed to `MEMORY_AND_DISK`
to match Scala in 2.0.
"""
self.is_cached = True
- self._jdf.cache()
+ self.persist(StorageLevel.MEMORY_AND_DISK)
Review comment:
@srowen
The difference in behavior is due to this code in `BlockManager.scala`.
```
// Put it in memory first, even if it also has useDisk set to true;
// We will drop it to disk later if the memory store can't hold it.
if (level.deserialized) {
memoryStore.putIteratorAsValues(blockId, iterator(), classTag)
match {
case Right(s) =>
size = s
case Left(iter) =>
// Not enough space to unroll this block; drop to disk if
applicable
if (level.useDisk) {
logWarning(s"Persisting block $blockId to disk instead.")
diskStore.put(blockId) { channel =>
val out = Channels.newOutputStream(channel)
serializerManager.dataSerializeStream(blockId, out,
iter)(classTag)
}
size = diskStore.getSize(blockId)
} else {
iteratorFromFailedMemoryStorePut = Some(iter)
}
}
} else { // !level.deserialized
memoryStore.putIteratorAsBytes(blockId, iterator(), classTag,
level.memoryMode) match {
case Right(s) =>
size = s
case Left(partiallySerializedValues) =>
// Not enough space to unroll this block; drop to disk if
applicable
if (level.useDisk) {
logWarning(s"Persisting block $blockId to disk instead.")
diskStore.put(blockId) { channel =>
val out = Channels.newOutputStream(channel)
partiallySerializedValues.finishWritingToStream(out)
}
size = diskStore.getSize(blockId)
} else {
iteratorFromFailedMemoryStorePut =
Some(partiallySerializedValues.valuesIterator)
}
}
}
```
When StorageLevel has `deserialized=true` `putIteratorAsValues` is used and
when `deserialized=false`, `putIteratorAsBytes` is used. Since data in python
is always serialized as mentioned by @ScrapCodes, we should always use
`putIteratorAsBytes`. To do this, we need to always keep StorageLevel with
`deserialized=false`. Calling pyspark's `persist()` ensures that.
@cloud-fan @HyukjinKwon Can you please comment if the approach taken by this
PR is correct?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
