Todd Lipcon has posted comments on this change. ( http://gerrit.cloudera.org:8080/8552 )
Change subject: KUDU-1454 [part 1]: update propagated timestamp on the driver ...................................................................... Patch Set 1: (6 comments) http://gerrit.cloudera.org:8080/#/c/8552/1//COMMIT_MSG Commit Message: http://gerrit.cloudera.org:8080/#/c/8552/1//COMMIT_MSG@9 PS1, Line 9: Currently, Saprk uses multiple clients simultaneously, possibly on typo http://gerrit.cloudera.org:8080/#/c/8552/1//COMMIT_MSG@16 PS1, Line 16: This patchs use Accumulator in Spark to properly update the propagated typo: patch http://gerrit.cloudera.org:8080/#/c/8552/1/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala File java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala: http://gerrit.cloudera.org:8080/#/c/8552/1/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala@58 PS1, Line 58: class TimestampAccumulator extends AccumulatorV2[Long, Long] { should this be marked private somehow? (I'm not good at Scala) I think if you defined this as: class TimestampAccumulator(var timestamp: Long = 0) extends ... ... then you'd automatically get an appropriate constructor which you could use more easily in copy() below http://gerrit.cloudera.org:8080/#/c/8552/1/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala@66 PS1, Line 66: override def copy(): AccumulatorV2[Long, Long] = { : val copy = new TimestampAccumulator() : copy.timestamp = timestamp : copy eg here you can just use = new TimestampAccumulator(timestamp) http://gerrit.cloudera.org:8080/#/c/8552/1/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala@86 PS1, Line 86: syncClient.updateLastPropagatedTimestamp(timestampAccumulator.value) it seems slightly odd that merge() has a side-effect here. Does this end up only called on the driver or also in each of the executors? http://gerrit.cloudera.org:8080/#/c/8552/1/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/KuduContext.scala@294 PS1, Line 294: timestampAccumulator.add(syncClient.getLastPropagatedTimestamp) should this be finally{} so that, even if there is an error, you can see all the rows you wrote prior to the error? also, how are we ensuring that the last-propagated timestamp gets propagated back _out_ to the executors for a further write? setting it in the executor client seems like it would only affect a read and not a later write. -- To view, visit http://gerrit.cloudera.org:8080/8552 To unsubscribe, visit http://gerrit.cloudera.org:8080/settings Gerrit-Project: kudu Gerrit-Branch: master Gerrit-MessageType: comment Gerrit-Change-Id: Id0a078ae8ebaa6a859be75c822879291192c5842 Gerrit-Change-Number: 8552 Gerrit-PatchSet: 1 Gerrit-Owner: Hao Hao <[email protected]> Gerrit-Reviewer: Dan Burkert <[email protected]> Gerrit-Reviewer: Kudu Jenkins Gerrit-Reviewer: Todd Lipcon <[email protected]> Gerrit-Comment-Date: Wed, 15 Nov 2017 23:01:04 +0000 Gerrit-HasComments: Yes
