Skip to content

[SPARK-18261][Structured Streaming] Add statistics to MemorySink for joining #15786

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
lw-lin wants to merge 2 commits into from
Closed

[SPARK-18261][Structured Streaming] Add statistics to MemorySink for joining #15786

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6dc2c9e
MemorySink statistics for joinintg
lw-lin Nov 5, 2016
a7a6f6b
Address comments
lw-lin Nov 6, 2016
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
Next Next commit
MemorySink statistics for joinintg
  • Loading branch information
@lw-lin
lw-lin committed Nov 5, 2016
commit 6dc2c9e18072756d51c4318b9d8a3e99fd4e2c4d
6 changes: 5 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/memory.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.encoders.encoderFor
import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.plans.logical.LeafNode
import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, Statistics}
import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.types.StructType
import org.apache.spark.util.Utils
Expand Down Expand Up @@ -212,4 +212,8 @@ class MemorySink(val schema: StructType, outputMode: OutputMode) extends Sink wi
*/
case class MemoryPlan(sink: MemorySink, output: Seq[Attribute]) extends LeafNode {
def this(sink: MemorySink) = this(sink, sink.schema.toAttributes)

private val sizePerRow = sink.schema.toAttributes.map(_.dataType.defaultSize).sum

override def statistics: Statistics = Statistics(sizePerRow * sink.allData.size)
Copy link
Contributor

@brkyvz brkyvz Nov 7, 2016

Choose a reason for hiding this comment

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

everywhere else, this is defined as a lazy val. Can you make it such that:

override lazy val statistics: Statistics  = {
  val sizePerRow = sink.schema.toAttributes.map(_.dataType.defaultSize).sum
  Statistics(sizeInBytes = sizePerRow * sink.allData.size)
}

Copy link
Contributor Author

@lw-lin lw-lin Nov 8, 2016

Choose a reason for hiding this comment

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

thanks @brkyvz. I'm afraid it should not be lazy val as the statistics should vary from batch to batch -- sink.allData.sizeshould vary from batch to batch. sizePerRow, however, does not vary from batch to batch, thus it's private val as it is now.

Copy link
Contributor

@brkyvz brkyvz Nov 8, 2016

Choose a reason for hiding this comment

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

oh yeah! you're right! good catch!

}
25 changes: 25 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/streaming/MemorySinkSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,6 +187,31 @@ class MemorySinkSuite extends StreamTest with BeforeAndAfter {
query.stop()
}

test("MemoryPlan statistics for joining") {
Copy link
Contributor

@brkyvz brkyvz Nov 5, 2016

Choose a reason for hiding this comment

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

it would also be nicer to just implement a test on the MemoryPlan itself, without needing to start a stream, purely on statistics calculation

val input = MemoryStream[Int]
val query = input.toDF()
.writeStream
.format("memory")
.queryName("memStream")
.start()

val memStream = spark.table("memStream").as[Int]

input.addData(1)
query.processAllAvailable()
checkDatasetUnorderly(
memStream.crossJoin(memStream.withColumnRenamed("value", "value2")).as[(Int, Int)],
(1, 1))

input.addData(2)
query.processAllAvailable()
checkDatasetUnorderly(
memStream.crossJoin(memStream.withColumnRenamed("value", "value2")).as[(Int, Int)],
(1, 1), (1, 2), (2, 1), (2, 2))

query.stop()
Copy link
Contributor

@brkyvz brkyvz Nov 5, 2016

Choose a reason for hiding this comment

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

can you put this in a try-finally block please?

}

ignore("stress test") {
// Ignore the stress test as it takes several minutes to run
(0 until 1000).foreach { _ =>
Expand Down