Skip to content

[SPARK-12091] [PYSPARK] Deprecate the JAVA-specific deserialized storage levels #10092

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
gatorsmile wants to merge 8 commits into from
Closed

[SPARK-12091] [PYSPARK] Deprecate the JAVA-specific deserialized storage levels #10092

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
bd7af48
Changed the default storage level of persist
gatorsmile Dec 2, 2015
d52deb5
Merge remote-tracking branch 'upstream/master' into persistStorageLevel
gatorsmile Dec 2, 2015
a6b7dd9
Removal of the JAVA-specific deserialized storage levels
gatorsmile Dec 3, 2015
0e074b6
Renaming MEMORY_ONLY_SER to MEMORY_ONLY
gatorsmile Dec 3, 2015
15e668e
Add deprecated notes for the four deprecated storage levels.
gatorsmile Dec 3, 2015
014a3a8
Add deprecated notes for the four deprecated storage levels.
gatorsmile Dec 3, 2015
fef7ada
change the deprecated version from 1.6 to 2.0
gatorsmile Dec 3, 2015
3864f87
update documents.
gatorsmile Dec 19, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions python/pyspark/rdd.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -220,18 +220,18 @@ def context(self):

def cache(self):
"""
Persist this RDD with the default storage level (C{MEMORY_ONLY_SER}).
Persist this RDD with the default storage level (C{MEMORY_ONLY}).
"""
self.is_cached = True
self.persist(StorageLevel.MEMORY_ONLY_SER)
self.persist(StorageLevel.MEMORY_ONLY)
return self

def persist(self, storageLevel=StorageLevel.MEMORY_ONLY_SER):
def persist(self, storageLevel=StorageLevel.MEMORY_ONLY):
"""
Set this RDD's storage level to persist its values across operations
after the first time it is computed. This can only be used to assign
a new storage level if the RDD does not have a storage level set yet.
If no storage level is specified defaults to (C{MEMORY_ONLY_SER}).
If no storage level is specified defaults to (C{MEMORY_ONLY}).

>>> rdd = sc.parallelize(["b", "a", "c"])
>>> rdd.persist().is_cached
Expand Down
6 changes: 3 additions & 3 deletions python/pyspark/sql/dataframe.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -371,18 +371,18 @@ def foreachPartition(self, f):

@since(1.3)
def cache(self):
""" Persists with the default storage level (C{MEMORY_ONLY_SER}).
""" Persists with the default storage level (C{MEMORY_ONLY}).
"""
self.is_cached = True
self._jdf.cache()
return self

@since(1.3)
def persist(self, storageLevel=StorageLevel.MEMORY_ONLY_SER):
def persist(self, storageLevel=StorageLevel.MEMORY_ONLY):
"""Sets the storage level to persist its values across operations
after the first time it is computed. This can only be used to assign
a new storage level if the RDD does not have a storage level set yet.
If no storage level is specified defaults to (C{MEMORY_ONLY_SER}).
If no storage level is specified defaults to (C{MEMORY_ONLY}).
"""
self.is_cached = True
javaStorageLevel = self._sc._getJavaStorageLevel(storageLevel)
Expand Down
18 changes: 8 additions & 10 deletions python/pyspark/storagelevel.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,10 @@ class StorageLevel(object):
"""
Flags for controlling the storage of an RDD. Each StorageLevel records whether to use memory,
whether to drop the RDD to disk if it falls out of memory, whether to keep the data in memory
in a serialized format, and whether to replicate the RDD partitions on multiple nodes.
Also contains static constants for some commonly used storage levels, such as MEMORY_ONLY.
in a JAVA-specific serialized format, and whether to replicate the RDD partitions on multiple
nodes. Also contains static constants for some commonly used storage levels, MEMORY_ONLY.
Since the data is always serialized on the Python side, all the constants use the serialized
formats.
"""

def __init__(self, useDisk, useMemory, useOffHeap, deserialized, replication=1):
Expand All @@ -49,12 +51,8 @@ def __str__(self):

StorageLevel.DISK_ONLY = StorageLevel(True, False, False, False)
StorageLevel.DISK_ONLY_2 = StorageLevel(True, False, False, False, 2)
StorageLevel.MEMORY_ONLY = StorageLevel(False, True, False, True)
StorageLevel.MEMORY_ONLY_2 = StorageLevel(False, True, False, True, 2)
StorageLevel.MEMORY_ONLY_SER = StorageLevel(False, True, False, False)
StorageLevel.MEMORY_ONLY_SER_2 = StorageLevel(False, True, False, False, 2)
StorageLevel.MEMORY_AND_DISK = StorageLevel(True, True, False, True)
StorageLevel.MEMORY_AND_DISK_2 = StorageLevel(True, True, False, True, 2)
StorageLevel.MEMORY_AND_DISK_SER = StorageLevel(True, True, False, False)
StorageLevel.MEMORY_AND_DISK_SER_2 = StorageLevel(True, True, False, False, 2)
StorageLevel.MEMORY_ONLY = StorageLevel(False, True, False, False)

@davies davies Dec 3, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removing these will break backward compatibility, I'd like to deprecate them, explain the difference between Python and Java (say records will always serialized in Python)

@gatorsmile gatorsmile Dec 3, 2015

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree! Just updated the codes with the deprecated notes. Trying to follow the existing PySpark style. Please check if they are good. : )

Not sure if this will be merged to 1.6. The note is still using 1.6. Thank you!

@davies davies Dec 3, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's too late for 1.6, and this change (API change) is good for 2.0, sounds good?

@gatorsmile gatorsmile Dec 3, 2015

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure. Just changed it. : )

StorageLevel.MEMORY_ONLY_2 = StorageLevel(False, True, False, False, 2)
StorageLevel.MEMORY_AND_DISK = StorageLevel(True, True, False, False)
StorageLevel.MEMORY_AND_DISK_2 = StorageLevel(True, True, False, False, 2)
StorageLevel.OFF_HEAP = StorageLevel(False, False, True, False, 1)
2 changes: 1 addition & 1 deletion python/pyspark/streaming/context.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -258,7 +258,7 @@ def checkpoint(self, directory):
"""
self._jssc.checkpoint(directory)

def socketTextStream(self, hostname, port, storageLevel=StorageLevel.MEMORY_AND_DISK_SER_2):
def socketTextStream(self, hostname, port, storageLevel=StorageLevel.MEMORY_AND_DISK_2):
"""
Create an input from TCP source hostname:port. Data is received using
a TCP socket and receive byte is interpreted as UTF8 encoded ``\\n`` delimited
Expand Down
4 changes: 2 additions & 2 deletions python/pyspark/streaming/dstream.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,10 +208,10 @@ def func(iterator):
def cache(self):
"""
Persist the RDDs of this DStream with the default storage level
(C{MEMORY_ONLY_SER}).
(C{MEMORY_ONLY}).
"""
self.is_cached = True
self.persist(StorageLevel.MEMORY_ONLY_SER)
self.persist(StorageLevel.MEMORY_ONLY)
return self

def persist(self, storageLevel):
Expand Down
4 changes: 2 additions & 2 deletions python/pyspark/streaming/flume.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ class FlumeUtils(object):

@staticmethod
def createStream(ssc, hostname, port,
storageLevel=StorageLevel.MEMORY_AND_DISK_SER_2,
storageLevel=StorageLevel.MEMORY_AND_DISK_2,
enableDecompression=False,
bodyDecoder=utf8_decoder):
"""
Expand Down Expand Up @@ -70,7 +70,7 @@ def createStream(ssc, hostname, port,

@staticmethod
def createPollingStream(ssc, addresses,
storageLevel=StorageLevel.MEMORY_AND_DISK_SER_2,
storageLevel=StorageLevel.MEMORY_AND_DISK_2,
maxBatchSize=1000,
parallelism=5,
bodyDecoder=utf8_decoder):
Expand Down
2 changes: 1 addition & 1 deletion python/pyspark/streaming/kafka.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ class KafkaUtils(object):

@staticmethod
def createStream(ssc, zkQuorum, groupId, topics, kafkaParams=None,
storageLevel=StorageLevel.MEMORY_AND_DISK_SER_2,
storageLevel=StorageLevel.MEMORY_AND_DISK_2,
keyDecoder=utf8_decoder, valueDecoder=utf8_decoder):
"""
Create an input stream that pulls messages from a Kafka Broker.
Expand Down
2 changes: 1 addition & 1 deletion python/pyspark/streaming/mqtt.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ class MQTTUtils(object):

@staticmethod
def createStream(ssc, brokerUrl, topic,
storageLevel=StorageLevel.MEMORY_AND_DISK_SER_2):
storageLevel=StorageLevel.MEMORY_AND_DISK_2):
"""
Create an input stream that pulls messages from a Mqtt Broker.

Expand Down