[SPARK-24525][SS] Provide an option to limit number of rows in a MemorySink #21559

mukulmurthy · 2018-06-13T20:59:30Z

What changes were proposed in this pull request?

Provide an option to limit number of rows in a MemorySink. Currently, MemorySink and MemorySinkV2 have unbounded size, meaning that if they're used on big data, they can OOM the stream. This change adds a maxMemorySinkRows option to limit how many rows MemorySink and MemorySinkV2 can hold. By default, they are still unbounded.

How was this patch tested?

Added new unit tests.

mukulmurthy · 2018-06-13T21:01:54Z

@jose-torres @brkyvz for review

brkyvz · 2018-06-13T21:42:35Z

ok to test

brkyvz · 2018-06-13T21:46:55Z

Jenkins add to whitelist

JoshRosen · 2018-06-13T22:00:24Z

jenkins add to whitelist

mukulmurthy · 2018-06-13T23:43:45Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/memory.scala

+  }
+
+  private def truncateRowsIfNeeded(rows: Array[Row], maxRows: Int, batchId: Long): Array[Row] = {
+    if (rows.length > maxRows) {


Also adding a check here to make sure maxRows >= 0. It shouldn't ever be negative, but doesn't hurt to safeguard.

SparkQA · 2018-06-14T01:35:53Z

Test build #91796 has finished for PR 21559 at commit f981cb8.

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

SparkQA · 2018-06-14T02:18:08Z

Test build #91799 has finished for PR 21559 at commit 4ab9bda.

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

jose-torres · 2018-06-14T17:56:53Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/memory.scala

+    numRows = 0
+  }
+
+  private def truncateRowsIfNeeded(rows: Array[Row], maxRows: Int, batchId: Long): Array[Row] = {


nit: I'd document that maxRows is the remaining row capacity, not the maximum row limit defined in the options. I got confused for a minute here.

jose-torres · 2018-06-14T17:57:53Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/memoryV2.scala

+    numRows = 0
+  }
+
+  private def truncateRowsIfNeeded(rows: Array[Row], maxRows: Int, batchId: Long): Array[Row] = {


Can this go in MemorySinkBase?

jose-torres · 2018-06-14T20:34:33Z

lgtm

brkyvz · 2018-06-13T23:44:22Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/memory.scala

+ * Companion object to MemorySinkBase.
+ */
+object MemorySinkBase {
+  val MAX_MEMORY_SINK_ROWS = "maxMemorySinkRows"


maxRows is sufficient

brkyvz · 2018-06-13T23:45:09Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/memory.scala

+  }
+}
+
+


nit: remove extra line

brkyvz · 2018-06-13T23:49:22Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/memory.scala

+   */
+  def getMemorySinkCapacity(options: DataSourceOptions): Option[Int] = {
+    val maxRows = options.getInt(MAX_MEMORY_SINK_ROWS, MAX_MEMORY_SINK_ROWS_DEFAULT)
+    if (maxRows >= 0) Some(maxRows) else None


Do you want to do if (maxRows >= 0) maxRows else Int.MaxValue - 10
We can't exceed runtime array max size anyway

brkyvz · 2018-06-13T23:50:25Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/memoryV2.scala

@@ -81,22 +84,35 @@ class MemorySinkV2 extends DataSourceV2 with StreamWriteSupport with MemorySinkB
    }.mkString("\n")
  }

-  def write(batchId: Long, outputMode: OutputMode, newRows: Array[Row]): Unit = {
+  def write(batchId: Long, outputMode: OutputMode, newRows: Array[Row], sinkCapacity: Option[Int])


nit: our style is more like

def write( batchId: Long, outputMode: OutputMode, newRows: Array[Row], sinkCapacity: Option[Int]): Unit = {

brkyvz · 2018-06-13T23:53:24Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/memoryV2.scala

+
+  private def truncateRowsIfNeeded(rows: Array[Row], maxRows: Int, batchId: Long): Array[Row] = {
+    if (rows.length > maxRows) {
+      logWarning(s"Truncating batch $batchId to $maxRows rows")


How does take behave with negative rows? Printing a warning message with negative values may be weird. I would also include the sink limit in the warning.

SparkQA · 2018-06-14T21:19:31Z

Test build #91861 has finished for PR 21559 at commit b2ef59c.

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

SparkQA · 2018-06-14T22:21:35Z

Test build #91864 has finished for PR 21559 at commit 25d6de1.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait MemorySinkBase extends BaseStreamingSink with Logging

SparkQA · 2018-06-15T01:06:29Z

Test build #91868 has finished for PR 21559 at commit e5b6175.

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

brkyvz

LGTM. Just two minor nits

brkyvz · 2018-06-15T16:31:24Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/memory.scala

+   * Gets the max number of rows a MemorySink should store. This number is based on the memory
+   * sink row limit if it is set. If not, there is no limit.
+   * @param options Options for writing from which we get the max rows option
+   * @return The maximum number of rows a memorySink should store, or None for no limit.


need to update docs

brkyvz · 2018-06-15T16:37:21Z

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/memory.scala

+      sinkLimit: Int,
+      batchId: Long): Array[Row] = {
+    if (rows.length > batchLimit && batchLimit >= 0) {
+      logWarning(s"Truncating batch $batchId to $batchLimit rows because of sink limit $sinkLimit")


nit: not sure if these sinks get used by Continuous processing too. If so I would rename batch to trigger version.

This piece is shared by MemorySink and MemorySinkV2, and the MemorySinkV2 (continuous processing) sink still calls them batches.

SparkQA · 2018-06-15T20:40:19Z

Test build #91926 has finished for PR 21559 at commit 0402b60.

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

brkyvz · 2018-06-15T20:52:09Z

Thanks! Merging to master!

…rySink Provide an option to limit number of rows in a MemorySink. Currently, MemorySink and MemorySinkV2 have unbounded size, meaning that if they're used on big data, they can OOM the stream. This change adds a maxMemorySinkRows option to limit how many rows MemorySink and MemorySinkV2 can hold. By default, they are still unbounded. Added new unit tests. Author: Mukul Murthy <mukul.murthy@databricks.com> Closes apache#21559 from mukulmurthy/SPARK-24525. Ref: LIHADOOP-48531 RB=1852593 G=superfriends-reviewers R=mshen,fli,latang,yezhou,zolin A=

mukulmurthy added 10 commits June 11, 2018 17:38

Add in logic to determine the max rows a sink can have

ac7eb2f

Make MemorySink and MemorySinkV2 respect row and byte limits

8dc89cc

Make tests compile

8ddf566

Make microbatch memory writer work with limits

d82c7d5

Test MemorySinkV2 with limits

7fefe87

Add MemorySink test with limit

58c5044

rename method

392f05f

Don't use byte limit, and log if we truncate rows

9097dd5

Update tests

a28fb38

minor refactor

f981cb8

mukulmurthy added 2 commits June 13, 2018 15:00

fixed indenting a bit

74d5b6b

make helper methods private

4ab9bda

mukulmurthy commented Jun 13, 2018

View reviewed changes

Add additional safeguard that we don't call take() on a negative number

b2ef59c

jose-torres reviewed Jun 14, 2018

View reviewed changes

Move truncate method to parent class

25d6de1

brkyvz reviewed Jun 14, 2018

View reviewed changes

Address Burak's comments

e5b6175

brkyvz approved these changes Jun 15, 2018

View reviewed changes

fix documentation

0402b60

asfgit closed this in e4fee39 Jun 15, 2018

mukulmurthy deleted the SPARK-24525 branch July 11, 2018 22:33

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-24525][SS] Provide an option to limit number of rows in a MemorySink #21559

[SPARK-24525][SS] Provide an option to limit number of rows in a MemorySink #21559

mukulmurthy commented Jun 13, 2018 •

edited

Loading

mukulmurthy commented Jun 13, 2018

brkyvz commented Jun 13, 2018

brkyvz commented Jun 13, 2018

JoshRosen commented Jun 13, 2018

mukulmurthy Jun 13, 2018

SparkQA commented Jun 14, 2018

SparkQA commented Jun 14, 2018

jose-torres Jun 14, 2018 •

edited

Loading

jose-torres Jun 14, 2018

jose-torres commented Jun 14, 2018

brkyvz Jun 13, 2018

brkyvz Jun 13, 2018

brkyvz Jun 13, 2018

brkyvz Jun 13, 2018

brkyvz Jun 13, 2018

SparkQA commented Jun 14, 2018

SparkQA commented Jun 14, 2018

SparkQA commented Jun 15, 2018

brkyvz left a comment

brkyvz Jun 15, 2018

brkyvz Jun 15, 2018

mukulmurthy Jun 15, 2018

SparkQA commented Jun 15, 2018

brkyvz commented Jun 15, 2018

[SPARK-24525][SS] Provide an option to limit number of rows in a MemorySink #21559

[SPARK-24525][SS] Provide an option to limit number of rows in a MemorySink #21559

Conversation

mukulmurthy commented Jun 13, 2018 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

mukulmurthy commented Jun 13, 2018

brkyvz commented Jun 13, 2018

brkyvz commented Jun 13, 2018

JoshRosen commented Jun 13, 2018

Choose a reason for hiding this comment

SparkQA commented Jun 14, 2018

SparkQA commented Jun 14, 2018

jose-torres Jun 14, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jose-torres commented Jun 14, 2018

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 Jun 14, 2018

SparkQA commented Jun 14, 2018

SparkQA commented Jun 15, 2018

brkyvz 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

SparkQA commented Jun 15, 2018

brkyvz commented Jun 15, 2018

mukulmurthy commented Jun 13, 2018 •

edited

Loading

jose-torres Jun 14, 2018 •

edited

Loading