[SPARK-22649][PYTHON][SQL] Adding localCheckpoint to Dataset API #19805

ferdonline · 2017-11-23T19:26:24Z

What changes were proposed in this pull request?

This change adds local checkpoint support to datasets and respective bind from Python Dataframe API.

If reliability requirements can be lowered to favor performance, as in cases of further quick transformations followed by a reliable save, localCheckpoints() fit very well.
Furthermore, at the moment Reliable checkpoints still incur double computation (see #9428)
In general it makes the API more complete as well.

How was this patch tested?

Python land quick use case:

>>> from time import sleep
>>> from pyspark.sql import types as T
>>> from pyspark.sql import functions as F

>>> def f(x):
    sleep(1)
    return x*2
   ...: 

>>> df1 = spark.range(30, numPartitions=6)
>>> df2 = df1.select(F.udf(f, T.LongType())("id"))

>>> %time _ = df2.collect()
CPU times: user 7.79 ms, sys: 5.84 ms, total: 13.6 ms                           
Wall time: 12.2 s

>>> %time df3 = df2.localCheckpoint()
CPU times: user 2.38 ms, sys: 2.3 ms, total: 4.68 ms                            
Wall time: 10.3 s

>>> %time _ = df3.collect()
CPU times: user 5.09 ms, sys: 410 µs, total: 5.5 ms
Wall time: 148 ms

>>> sc.setCheckpointDir(".")
>>> %time df3 = df2.checkpoint()
CPU times: user 4.04 ms, sys: 1.63 ms, total: 5.67 ms                           
Wall time: 20.3 s

ferdonline · 2017-11-24T09:28:58Z

retest this please

ferdonline · 2017-11-24T14:10:15Z

cc @andrewor14 since I believe you know this part of spark pretty well, maybe you could help me integrating this.
Any idea why Jenkins didn't start the testing?

felixcheung · 2017-11-26T20:03:56Z

Jenkins, test this please

felixcheung · 2017-11-26T20:04:11Z

could you update the title to add [PYTHON]

felixcheung · 2017-11-26T20:08:23Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

 *
 * @group basic
 * @since 2.1.0
 */
 @Experimental
 @InterfaceStability.Evolving
- def checkpoint(eager: Boolean): Dataset[T] = {
+ def _checkpoint(eager: Boolean, local: Boolean = false): Dataset[T] = {


I don't think this is right - this is still a public API and it's not the convention here.
Add it as a private instead

Sounds good. Regarding the naming, I'm not sure adding an underscore is a good choice. Let me know if it should be changed.

felixcheung · 2017-11-26T20:09:02Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -524,22 +524,41 @@ class Dataset[T] private[sql](
 */
 @Experimental
 @InterfaceStability.Evolving
- def checkpoint(): Dataset[T] = checkpoint(eager = true)
+ def checkpoint(eager: Boolean = true): Dataset[T] = _checkpoint(eager = eager)


I think we should keep this signature - there's already a def checkpoint(eager: Boolean = true) below

I always try to avoid duplication of code, and with docs this takes ~10 lines for nothing - I believe a function with a default parameter is as readable as the function without the parameter. But please let me know if it goes against the code style. Thanks

No, it's going to break Java API compatibility. Default value does not work with Java.

felixcheung · 2017-11-26T20:09:22Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

+ * plan may grow exponentially. Local checkpoints are written to executor storage and despite
+ * potentially faster they are unreliable and may compromise job completion.
+ *
+ * @group basic


add @since

SparkQA · 2017-11-26T20:09:34Z

Test build #84196 has finished for PR 19805 at commit abe03ab.

This patch fails Python style tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2017-11-26T20:50:56Z

cc @zsxwing @liancheng

ferdonline · 2017-11-27T09:35:25Z

Thanks for you review. I'm working on the changes

ferdonline · 2017-11-28T23:36:03Z

I think I fixed everything that was suggested. Thanks for the input.
If someone could enable the tests and review would be great!

ueshin · 2017-11-29T04:45:45Z

ok to test

ueshin · 2017-11-29T04:50:22Z

@ferdonline Could you file a JIRA issue and add the id to the title like [SPARK-xxx][PYTHON][SQL] ...?

ueshin

I left some minor comments. Thanks!

ueshin · 2017-11-29T04:51:04Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -518,13 +518,12 @@ class Dataset[T] private[sql](
 * the logical plan of this Dataset, which is especially useful in iterative algorithms where the
 * plan may grow exponentially. It will be saved to files inside the checkpoint
 * directory set with `SparkContext#setCheckpointDir`.
- *


Maybe we should revert this back?

ueshin · 2017-11-29T04:52:19Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -537,9 +536,55 @@ class Dataset[T] private[sql](
 */
 @Experimental
 @InterfaceStability.Evolving
- def checkpoint(eager: Boolean): Dataset[T] = {
+ def checkpoint(eager: Boolean = true): Dataset[T] = _checkpoint(eager = eager)


We don't need the default value for eager here.

ueshin · 2017-11-29T04:52:45Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

+ */
+ @Experimental
+ @InterfaceStability.Evolving
+ def localCheckpoint(eager: Boolean = true): Dataset[T] = _checkpoint(eager = eager, local = true)


ueshin · 2017-11-29T04:58:35Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

+ */
+ @Experimental
+ @InterfaceStability.Evolving
+ private[sql] def _checkpoint(eager: Boolean, local: Boolean = false): Dataset[T] = {


We should make this private?

I changed according to the first review. I also agree to be private so that "there's only one way of doing it" when using the API. But if you have a strong feeling I can surely change

I guess we have 2 options here:

expose def checkpoint(eager: Boolean, local: Boolean): Dataset[T] as public, which can be used similar to localCheckpoint.

make def _checkpoint(eager: Boolean, local: Boolean = false): Dataset[T] private to be used only from the public APIs.

and I'm afraid the current one is not good anyway.

I'd prefer the second option but I don't have a strong feeling.
cc @felixcheung @HyukjinKwon

Sounds good to me to remove default param values, keeping private. Let me know

private checkpoint makes sense to me too. i think we don't need the underscore though as it's going to have private ahead and strictly it might be redundant to have the leading underscore to note it's private is.

Alright, so I remove both default values, otherwise we end up with a colliding signature.

SparkQA · 2017-11-29T07:31:09Z

Test build #84282 has finished for PR 19805 at commit 59b5562.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-11-29T12:01:01Z

Test build #84289 has finished for PR 19805 at commit c5f1b2c.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-11-29T17:26:19Z

Test build #84299 has finished for PR 19805 at commit 54b7f33.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HyukjinKwon · 2017-11-30T01:38:33Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

+ */
+ @Experimental
+ @InterfaceStability.Evolving
+ private[sql] def checkpoint(eager: Boolean, local: Boolean): Dataset[T] = {


Hm, can we just do this private def checkpoint instead of private[sql]? Also, I don't think it needs those @Experimental, @InterfaceStability.Evolving, @since 2.3.0 and @group basic.

SparkQA · 2017-11-30T12:35:58Z

Test build #84334 has finished for PR 19805 at commit c743c34.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2017-12-01T00:30:47Z

Will review it this weekend.

gatorsmile · 2017-12-02T07:32:40Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -537,9 +537,48 @@ class Dataset[T] private[sql](
 */
 @Experimental
 @InterfaceStability.Evolving
- def checkpoint(eager: Boolean): Dataset[T] = {


Could you check the test case of def checkpoint? At least we need to add a test case.

I can try to create a test to localCheckpoint based on the one for checkpoint, but I'm not very familiar with Scala and the Spark scala API, so currently I don't feel at ease to create a meaningful test. Would anybody be up to add one?

So we already test checkpoint in DatasetSuite

SparkQA · 2017-12-11T14:16:22Z

Test build #84706 has finished for PR 19805 at commit 9beb375.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-12-11T17:31:12Z

Test build #84709 has finished for PR 19805 at commit da34c4a.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds no public classes.

…_localCheckpoint

SparkQA · 2017-12-11T20:23:37Z

Test build #84720 has finished for PR 19805 at commit 7532fc4.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds no public classes.

ferdonline · 2017-12-12T13:25:15Z

Existing checkpoint tests applied to localCheckpoint as well, all working well. Please verify.
I'm getting an unrelated fail in SparkR, did anything change n the build system?

gatorsmile · 2017-12-17T08:01:35Z

sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala

+ }
+ }
+ }
+ else {


-> } else {

gatorsmile · 2017-12-17T08:03:08Z

retest this please

gatorsmile · 2017-12-17T08:05:20Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

+ if (reliableCheckpoint) {
+ internalRdd.checkpoint()
+ } else {
+ internalRdd.localCheckpoint()


Could you also issue a logWarning message here to indicate the checkpoint is not reliable? This call is a potential issue when users using AWS EC2 Spot instances.

Hi. Thanks for the review.
From the point of view of the user being aware he's doing a local checkpoint we already force him to use localCheckpoint() (the generic checkpoint is private)
If we should warn users about the potential issues with localCheckpoint() shouldn't we do it in the RDD API, so that users are always warned?

cc @zsxwing

SparkQA · 2017-12-17T10:54:31Z

Test build #85017 has finished for PR 19805 at commit 7532fc4.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-12-17T18:39:02Z

Test build #85027 has finished for PR 19805 at commit 45f4bf5.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2017-12-19T22:50:00Z

retest this please

SparkQA · 2017-12-20T01:38:54Z

Test build #85139 has finished for PR 19805 at commit 45f4bf5.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2017-12-20T04:46:45Z

LGTM

Thanks! Merged to master.

Adding localCheckpoint to Dataframe API

abe03ab

ferdonline changed the title ~~Adding localCheckpoint to Dataframe API~~ [SQL] Adding localCheckpoint to Dataset API Nov 24, 2017

felixcheung reviewed Nov 26, 2017

View reviewed changes

ferdonline changed the title ~~[SQL] Adding localCheckpoint to Dataset API~~ [Python][SQL] Adding localCheckpoint to Dataset API Nov 27, 2017

ferdonline changed the title ~~[Python][SQL] Adding localCheckpoint to Dataset API~~ [PYTHON][SQL] Adding localCheckpoint to Dataset API Nov 27, 2017

ferdonline added 2 commits November 27, 2017 11:05

[SQL] localCheckpoint: Style fixes

aa70ce6

[SQL] localCheckpoint() adding no-param declaration

59b5562

ueshin reviewed Nov 29, 2017

View reviewed changes

ferdonline changed the title ~~[PYTHON][SQL] Adding localCheckpoint to Dataset API~~ [SPARK-22649][PYTHON][SQL] Adding localCheckpoint to Dataset API Nov 29, 2017

Default args removed in methods having equivalent signature

c5f1b2c

No default values

54b7f33

HyukjinKwon reviewed Nov 30, 2017

View reviewed changes

checkpoint tags and docstring

c743c34

gatorsmile reviewed Dec 2, 2017

View reviewed changes

Adding tests

9beb375

fix

da34c4a

Merge branch 'master' of github.com:apache/spark into feature_dataset…

7532fc4

…_localCheckpoint

gatorsmile reviewed Dec 17, 2017

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala Outdated

}

}

}

else {

Copy link

Member

gatorsmile Dec 17, 2017

Choose a reason for hiding this comment

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

-> } else {

gatorsmile reviewed Dec 17, 2017

View reviewed changes

style fix

45f4bf5

asfgit closed this in 13268a5 Dec 20, 2017

[SPARK-22649][PYTHON][SQL] Adding localCheckpoint to Dataset API #19805

[SPARK-22649][PYTHON][SQL] Adding localCheckpoint to Dataset API #19805

Conversation

ferdonline commented Nov 23, 2017 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

ferdonline commented Nov 24, 2017

ferdonline commented Nov 24, 2017

felixcheung commented Nov 26, 2017

felixcheung commented Nov 26, 2017

felixcheung Nov 26, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

felixcheung Nov 26, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Nov 26, 2017

gatorsmile commented Nov 26, 2017

ferdonline commented Nov 27, 2017

ferdonline commented Nov 28, 2017 • edited Loading

ueshin commented Nov 29, 2017

ueshin commented Nov 29, 2017

ueshin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Nov 29, 2017

SparkQA commented Nov 29, 2017

SparkQA commented Nov 29, 2017

Choose a reason for hiding this comment

SparkQA commented Nov 30, 2017

gatorsmile commented Dec 1, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 11, 2017

SparkQA commented Dec 11, 2017

SparkQA commented Dec 11, 2017

ferdonline commented Dec 12, 2017

Choose a reason for hiding this comment

gatorsmile commented Dec 17, 2017

gatorsmile Dec 17, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 17, 2017

SparkQA commented Dec 17, 2017

gatorsmile commented Dec 19, 2017

SparkQA commented Dec 20, 2017

gatorsmile commented Dec 20, 2017

ferdonline commented Nov 23, 2017 •

edited

Loading

felixcheung Nov 26, 2017 •

edited

Loading

felixcheung Nov 26, 2017 •

edited

Loading

ferdonline commented Nov 28, 2017 •

edited

Loading

gatorsmile Dec 17, 2017 •

edited

Loading