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

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
Prev Previous commit
Address comments
  • Loading branch information
@lw-lin
lw-lin committed Nov 6, 2016
commit a7a6f6bbe3d7b4245a1a568cd16332ce6af61e73
31 changes: 11 additions & 20 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,29 +187,20 @@ class MemorySinkSuite extends StreamTest with BeforeAndAfter {
query.stop()
}

test("MemoryPlan statistics for joining") {
val input = MemoryStream[Int]
val query = input.toDF()
.writeStream
.format("memory")
.queryName("memStream")
.start()

val memStream = spark.table("memStream").as[Int]
test("MemoryPlan statistics") {
implicit val schema = new StructType().add(new StructField("value", IntegerType))
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.

thanks! This will be much faster than a test where we actually join DataFrames.

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.

happy to do this

val sink = new MemorySink(schema, InternalOutputModes.Append)
val plan = new MemoryPlan(sink)

input.addData(1)
query.processAllAvailable()
checkDatasetUnorderly(
memStream.crossJoin(memStream.withColumnRenamed("value", "value2")).as[(Int, Int)],
(1, 1))
// Before adding data, check output
checkAnswer(sink.allData, Seq.empty)
assert(plan.statistics.sizeInBytes === 0)

input.addData(2)
query.processAllAvailable()
checkDatasetUnorderly(
memStream.crossJoin(memStream.withColumnRenamed("value", "value2")).as[(Int, Int)],
(1, 1), (1, 2), (2, 1), (2, 2))
sink.addBatch(0, 1 to 3)
assert(plan.statistics.sizeInBytes === 12)

query.stop()
sink.addBatch(1, 4 to 6)
assert(plan.statistics.sizeInBytes === 24)
}

ignore("stress test") {
Expand Down