Skip to content

[SPARK-19800][SS][WIP] Implement one kind of streaming sampling - reservoir sampling - #17141

Closed
uncleGen wants to merge 6 commits into
apache:masterfrom
uncleGen:sampling
Closed

[SPARK-19800][SS][WIP] Implement one kind of streaming sampling - reservoir sampling#17141
uncleGen wants to merge 6 commits into
apache:masterfrom
uncleGen:sampling

Conversation

@uncleGen

@uncleGenuncleGen commented Mar 2, 2017•
edited
Loading

Copy link
Copy Markdown
Contributor

What changes were proposed in this pull request?

This pr adds a special streaming sample operator to support sample. It adds a new evolving operator reservoir, and introduce an new logical plan ReservoirSample and two physical plan StreamingReservoirSampleExec and ReservoirSampleExec.

The following cases are supported:

  • batch table reservoir sampling
  • stream table reservoir sampling with/without aggregation and watermark in Update/Complete output mode

Not supported cases:

  • reservoir sampling in Append output mode (No meaning)

Followups:

  • move reservoir into sample operator

How was this patch tested?

add new unit tests.

@srowen

srowen commented Mar 2, 2017

Copy link
Copy Markdown
Member

Why does this need to be in Spark?

@SparkQA

SparkQA commented Mar 2, 2017

Copy link
Copy Markdown

Test build #73778 has finished for PR 17141 at commit 288c124.

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

@uncleGen

uncleGen commented Mar 3, 2017•
edited
Loading

Copy link
Copy Markdown
ContributorAuthor

@srowen There are some unsupported operator for Structured Streaming. You can view here: https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala#L184
This pr adds support for sample operator, more exactly reservoir operator. But next step, I will try to combine reservoir into sample

@uncleGen

uncleGen commented Mar 3, 2017

Copy link
Copy Markdown
ContributorAuthor

cc @zsxwing and @tdas

uncleGen
uncleGen commented Mar 3, 2017
View reviewed changes
Comment threadsql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
private val enc = Encoders.STRING.asInstanceOf[ExpressionEncoder[String]]
private val NUM_RECORDS_IN_PARTITION = enc.toRow("NUM_RECORDS_IN_PARTITION")
.asInstanceOf[UnsafeRow]

@uncleGenuncleGenMar 3, 2017

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

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

NUM_RECORDS_IN_PARTITION calculate the total number of records in current partiton, and update at the end of sample.

Comment threadsql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
UnsafeProjection.create(withSumFieldTypes).apply(InternalRow.fromSeq(
new JoinedRow(kv._2, numRecordsTillNow)
.toSeq(withSumFieldTypes)))
}), {})

@uncleGenuncleGenMar 3, 2017

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

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

Here, we transfer the row to (row, numRecordsTillNow), and numRecordsTillNow is used to calculate the weight of item.

Comment threadsql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
.map(update => {
UnsafeProjection.create(withSumFieldTypes).apply(InternalRow.fromSeq(
new JoinedRow(update.value, numRecordsTillNow)
.toSeq(withSumFieldTypes)))

@uncleGenuncleGenMar 3, 2017

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

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

same

Comment threadsql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
.apply(InternalRow.fromSeq(row.toSeq(fieldTypes)))
).iterator
})
}

@uncleGenuncleGenMar 3, 2017

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

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

here, we do once global weight reservoir sampling.

Comment threadsql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
store.put(replacementIdx, r.asInstanceOf[UnsafeRow])
}
}
}

@uncleGenuncleGenMar 3, 2017

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

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

In partiton, we just need to do once normal (without weight) reservoir sampling.

Comment threadsql/core/src/test/scala/org/apache/spark/sql/streaming/ReservoirSampleSuit.scala
.groupBy("value").count()
assert(df.count() == 3, "")
}
}

@uncleGenuncleGenMar 3, 2017

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

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

new unit test needs to be improved.

@uncleGen

uncleGen commented Mar 7, 2017

Copy link
Copy Markdown
ContributorAuthor

ping @tdas and @zsxwing

@SparkQA

SparkQA commented Mar 9, 2017

Copy link
Copy Markdown

Test build #74238 has finished for PR 17141 at commit c4008cd.

  • This patch passes all tests.
  • This patch merges cleanly.
  • This patch adds the following public classes (experimental):
  • public static class LongWrapper
  • public static class IntWrapper
  • case class ResolveInlineTables(conf: CatalystConf) extends Rule[LogicalPlan]
  • case class CostBasedJoinReorder(conf: CatalystConf) extends Rule[LogicalPlan] with PredicateHelper
  • case class JoinPlan(itemIds: Set[Int], plan: LogicalPlan, joinConds: Set[Expression], cost: Cost)
  • case class Cost(rows: BigInt, size: BigInt)
  • abstract class RepartitionOperation extends UnaryNode
  • case class FlatMapGroupsWithState(
  • class CSVOptions(
  • class UnivocityParser(
  • trait WatermarkSupport extends UnaryExecNode
  • case class FlatMapGroupsWithStateExec(

@SparkQA

SparkQA commented Mar 20, 2017

Copy link
Copy Markdown

Test build #74841 has finished for PR 17141 at commit 1ddb82e.

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

@SparkQA

SparkQA commented Mar 20, 2017

Copy link
Copy Markdown

Test build #74843 has finished for PR 17141 at commit 02d44aa.

  • This patch passes all tests.
  • This patch merges cleanly.
  • This patch adds the following public classes (experimental):
  • case class ReservoirSampleExec(reservoirSize: Int, child: SparkPlan) extends UnaryExecNode

@jiangxb1987

jiangxb1987 commented May 24, 2017

Copy link
Copy Markdown
Contributor

Is anyone still working on this or this could be closed? @uncleGen@zsxwing

@jiangxb1987jiangxb1987 mentioned this pull request Jun 7, 2017
Closed
@asfgitasfgit closed this in b771fedJun 8, 2017
zifeif2 pushed a commit to zifeif2/spark that referenced this pull request Nov 22, 2025
@HyukjinKwon@zifeif2
[INFRA] Close stale PRs
ce94f63
# What changes were proposed in this pull request?
This PR proposes to close stale PRs, mostly the same instances with apache#18017Closesapache#11459Closesapache#13833Closesapache#13720Closesapache#12506Closesapache#12456Closesapache#12252Closesapache#17689Closesapache#17791Closesapache#18163Closesapache#17640Closesapache#17926Closesapache#18163Closesapache#12506Closesapache#18044Closesapache#14036Closesapache#15831Closesapache#14461Closesapache#17638Closesapache#18222
Added:
Closesapache#18045Closesapache#18061Closesapache#18010Closesapache#18041Closesapache#18124Closesapache#18130Closesapache#12217
Added:
Closesapache#16291Closesapache#17480Closesapache#14995
Added:
Closesapache#12835Closesapache#17141
## How was this patch tested?
N/A
Author: hyukjinkwon <gurwls223@gmail.com>
Closesapache#18223 from HyukjinKwon/close-stale-prs.
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

4 participants

@uncleGen@srowen@SparkQA@jiangxb1987