You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by uncleGen <gi...@git.apache.org> on 2017/03/02 16:39:39 UTC
[GitHub] spark pull request #17141: [SPARK-19800][SS][WIP] Implement one kind of stre...
GitHub user uncleGen opened a pull request:
https://github.com/apache/spark/pull/17141
[SPARK-19800][SS][WIP] Implement one kind of streaming sampling - reservoir sampling
## What changes were proposed in this pull request?
This pr adds a special streaming sample operator to support `sample`. It add a new evolving operator `reservoir`, and introduce 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
Followups:
- move `reservoir` into `sample` operator
## How was this patch tested?
add new unit tests.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/uncleGen/spark sampling
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/17141.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #17141
----
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17141: [SPARK-19800][SS][WIP] Implement one kind of stre...
Posted by uncleGen <gi...@git.apache.org>.
Github user uncleGen commented on a diff in the pull request:
https://github.com/apache/spark/pull/17141#discussion_r104073516
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala ---
@@ -397,3 +402,110 @@ object StreamingDeduplicateExec {
private val EMPTY_ROW =
UnsafeProjection.create(Array[DataType](NullType)).apply(InternalRow.apply(null))
}
+
+/**
+ * Physical operator for executing streaming Sampling.
+ *
+ * @param k random sample k elements.
+ */
+case class StreamingReservoirSampleExec(
+ keyExpressions: Seq[Attribute],
+ child: SparkPlan,
+ k: Int,
+ stateId: Option[OperatorStateId] = None,
+ eventTimeWatermark: Option[Long] = None,
+ outputMode: Option[OutputMode] = None)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ override def requiredChildDistribution: Seq[Distribution] =
+ ClusteredDistribution(keyExpressions) :: Nil
+
+ private val enc = Encoders.STRING.asInstanceOf[ExpressionEncoder[String]]
+ private val NUM_RECORDS_IN_PARTITION = enc.toRow("NUM_RECORDS_IN_PARTITION")
+ .asInstanceOf[UnsafeRow]
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ metrics
+ val fieldTypes = (keyExpressions.map(_.dataType) ++ Seq(LongType)).toArray
+ val withSumFieldTypes = (keyExpressions.map(_.dataType) ++ Seq(LongType)).toArray
+
+ child.execute().mapPartitionsWithStateStore(
+ getStateId.checkpointLocation,
+ getStateId.operatorId,
+ getStateId.batchId,
+ keyExpressions.toStructType,
+ child.output.toStructType,
+ sqlContext.sessionState,
+ Some(sqlContext.streams.stateStoreCoordinator)) { (store, iter) =>
+
+ val numRecordsInPart = store.get(NUM_RECORDS_IN_PARTITION).map(value => {
+ value.get(0, LongType).asInstanceOf[Long]
+ }).getOrElse(0L)
+
+ val seed = Random.nextLong()
+ val rand = new XORShiftRandom(seed)
+ var numSamples = numRecordsInPart
+ var count = 0
+
+ val baseIterator = watermarkPredicate match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ baseIterator.foreach { r =>
+ count += 1
+ if (numSamples < k) {
+ numSamples += 1
+ store.put(enc.toRow(numSamples.toString).asInstanceOf[UnsafeRow],
+ r.asInstanceOf[UnsafeRow])
+ } else {
+ val randomIdx = (rand.nextDouble() * (numRecordsInPart + count)).toLong
+ if (randomIdx <= k) {
+ val replacementIdx = enc.toRow(randomIdx.toString).asInstanceOf[UnsafeRow]
+ store.put(replacementIdx, r.asInstanceOf[UnsafeRow])
+ }
+ }
+ }
+
+ val numRecordsTillNow = UnsafeProjection.create(Array[DataType](LongType))
+ .apply(InternalRow.apply(numRecordsInPart + count))
+ store.put(NUM_RECORDS_IN_PARTITION, numRecordsTillNow)
+ store.commit()
+
+ outputMode match {
+ case Some(Complete) =>
+ CompletionIterator[InternalRow, Iterator[InternalRow]](
+ store.iterator().filter(kv => {
+ !kv._1.asInstanceOf[UnsafeRow].equals(NUM_RECORDS_IN_PARTITION)
+ }).map(kv => {
+ UnsafeProjection.create(withSumFieldTypes).apply(InternalRow.fromSeq(
+ new JoinedRow(kv._2, numRecordsTillNow)
+ .toSeq(withSumFieldTypes)))
+ }), {})
--- End diff --
Here, we transfer the row to (row, numRecordsTillNow), and `numRecordsTillNow` is used to calculate the weight of item.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17141: [SPARK-19800][SS][WIP] Implement one kind of stre...
Posted by uncleGen <gi...@git.apache.org>.
Github user uncleGen commented on a diff in the pull request:
https://github.com/apache/spark/pull/17141#discussion_r104073674
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala ---
@@ -397,3 +402,110 @@ object StreamingDeduplicateExec {
private val EMPTY_ROW =
UnsafeProjection.create(Array[DataType](NullType)).apply(InternalRow.apply(null))
}
+
+/**
+ * Physical operator for executing streaming Sampling.
+ *
+ * @param k random sample k elements.
+ */
+case class StreamingReservoirSampleExec(
+ keyExpressions: Seq[Attribute],
+ child: SparkPlan,
+ k: Int,
+ stateId: Option[OperatorStateId] = None,
+ eventTimeWatermark: Option[Long] = None,
+ outputMode: Option[OutputMode] = None)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ override def requiredChildDistribution: Seq[Distribution] =
+ ClusteredDistribution(keyExpressions) :: Nil
+
+ private val enc = Encoders.STRING.asInstanceOf[ExpressionEncoder[String]]
+ private val NUM_RECORDS_IN_PARTITION = enc.toRow("NUM_RECORDS_IN_PARTITION")
+ .asInstanceOf[UnsafeRow]
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ metrics
+ val fieldTypes = (keyExpressions.map(_.dataType) ++ Seq(LongType)).toArray
+ val withSumFieldTypes = (keyExpressions.map(_.dataType) ++ Seq(LongType)).toArray
+
+ child.execute().mapPartitionsWithStateStore(
+ getStateId.checkpointLocation,
+ getStateId.operatorId,
+ getStateId.batchId,
+ keyExpressions.toStructType,
+ child.output.toStructType,
+ sqlContext.sessionState,
+ Some(sqlContext.streams.stateStoreCoordinator)) { (store, iter) =>
+
+ val numRecordsInPart = store.get(NUM_RECORDS_IN_PARTITION).map(value => {
+ value.get(0, LongType).asInstanceOf[Long]
+ }).getOrElse(0L)
+
+ val seed = Random.nextLong()
+ val rand = new XORShiftRandom(seed)
+ var numSamples = numRecordsInPart
+ var count = 0
+
+ val baseIterator = watermarkPredicate match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ baseIterator.foreach { r =>
+ count += 1
+ if (numSamples < k) {
+ numSamples += 1
+ store.put(enc.toRow(numSamples.toString).asInstanceOf[UnsafeRow],
+ r.asInstanceOf[UnsafeRow])
+ } else {
+ val randomIdx = (rand.nextDouble() * (numRecordsInPart + count)).toLong
+ if (randomIdx <= k) {
+ val replacementIdx = enc.toRow(randomIdx.toString).asInstanceOf[UnsafeRow]
+ store.put(replacementIdx, r.asInstanceOf[UnsafeRow])
+ }
+ }
+ }
--- End diff --
In partiton, we just need to do once normal (without weight) reservoir sampling.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17141: [SPARK-19800][SS][WIP] Implement one kind of stre...
Posted by uncleGen <gi...@git.apache.org>.
Github user uncleGen commented on a diff in the pull request:
https://github.com/apache/spark/pull/17141#discussion_r104073535
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala ---
@@ -397,3 +402,110 @@ object StreamingDeduplicateExec {
private val EMPTY_ROW =
UnsafeProjection.create(Array[DataType](NullType)).apply(InternalRow.apply(null))
}
+
+/**
+ * Physical operator for executing streaming Sampling.
+ *
+ * @param k random sample k elements.
+ */
+case class StreamingReservoirSampleExec(
+ keyExpressions: Seq[Attribute],
+ child: SparkPlan,
+ k: Int,
+ stateId: Option[OperatorStateId] = None,
+ eventTimeWatermark: Option[Long] = None,
+ outputMode: Option[OutputMode] = None)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ override def requiredChildDistribution: Seq[Distribution] =
+ ClusteredDistribution(keyExpressions) :: Nil
+
+ private val enc = Encoders.STRING.asInstanceOf[ExpressionEncoder[String]]
+ private val NUM_RECORDS_IN_PARTITION = enc.toRow("NUM_RECORDS_IN_PARTITION")
+ .asInstanceOf[UnsafeRow]
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ metrics
+ val fieldTypes = (keyExpressions.map(_.dataType) ++ Seq(LongType)).toArray
+ val withSumFieldTypes = (keyExpressions.map(_.dataType) ++ Seq(LongType)).toArray
+
+ child.execute().mapPartitionsWithStateStore(
+ getStateId.checkpointLocation,
+ getStateId.operatorId,
+ getStateId.batchId,
+ keyExpressions.toStructType,
+ child.output.toStructType,
+ sqlContext.sessionState,
+ Some(sqlContext.streams.stateStoreCoordinator)) { (store, iter) =>
+
+ val numRecordsInPart = store.get(NUM_RECORDS_IN_PARTITION).map(value => {
+ value.get(0, LongType).asInstanceOf[Long]
+ }).getOrElse(0L)
+
+ val seed = Random.nextLong()
+ val rand = new XORShiftRandom(seed)
+ var numSamples = numRecordsInPart
+ var count = 0
+
+ val baseIterator = watermarkPredicate match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ baseIterator.foreach { r =>
+ count += 1
+ if (numSamples < k) {
+ numSamples += 1
+ store.put(enc.toRow(numSamples.toString).asInstanceOf[UnsafeRow],
+ r.asInstanceOf[UnsafeRow])
+ } else {
+ val randomIdx = (rand.nextDouble() * (numRecordsInPart + count)).toLong
+ if (randomIdx <= k) {
+ val replacementIdx = enc.toRow(randomIdx.toString).asInstanceOf[UnsafeRow]
+ store.put(replacementIdx, r.asInstanceOf[UnsafeRow])
+ }
+ }
+ }
+
+ val numRecordsTillNow = UnsafeProjection.create(Array[DataType](LongType))
+ .apply(InternalRow.apply(numRecordsInPart + count))
+ store.put(NUM_RECORDS_IN_PARTITION, numRecordsTillNow)
+ store.commit()
+
+ outputMode match {
+ case Some(Complete) =>
+ CompletionIterator[InternalRow, Iterator[InternalRow]](
+ store.iterator().filter(kv => {
+ !kv._1.asInstanceOf[UnsafeRow].equals(NUM_RECORDS_IN_PARTITION)
+ }).map(kv => {
+ UnsafeProjection.create(withSumFieldTypes).apply(InternalRow.fromSeq(
+ new JoinedRow(kv._2, numRecordsTillNow)
+ .toSeq(withSumFieldTypes)))
+ }), {})
+ case Some(Update) =>
+ CompletionIterator[InternalRow, Iterator[InternalRow]](
+ store.updates()
+ .filter(update => !update.key.equals(NUM_RECORDS_IN_PARTITION))
+ .map(update => {
+ UnsafeProjection.create(withSumFieldTypes).apply(InternalRow.fromSeq(
+ new JoinedRow(update.value, numRecordsTillNow)
+ .toSeq(withSumFieldTypes)))
--- End diff --
same
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by jiangxb1987 <gi...@git.apache.org>.
Github user jiangxb1987 commented on the issue:
https://github.com/apache/spark/pull/17141
Is anyone still working on this or this could be closed? @uncleGen @zsxwing
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/17141
**[Test build #73778 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/73778/testReport)** for PR 17141 at commit [`288c124`](https://github.com/apache/spark/commit/288c124731901c0506833da05119fc131d1bf20e).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/17141
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/74238/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17141: [SPARK-19800][SS][WIP] Implement one kind of stre...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/17141
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/17141
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/74843/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/17141
**[Test build #74843 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/74843/testReport)** for PR 17141 at commit [`02d44aa`](https://github.com/apache/spark/commit/02d44aa06f025dc1d69a7abbcf59691ce7ee0e4e).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/17141
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/17141
**[Test build #74841 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/74841/testReport)** for PR 17141 at commit [`1ddb82e`](https://github.com/apache/spark/commit/1ddb82e5a5aaadebd28080d334d58d082a125da8).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by uncleGen <gi...@git.apache.org>.
Github user uncleGen commented on the issue:
https://github.com/apache/spark/pull/17141
cc @zsxwing and @tdas
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17141: [SPARK-19800][SS][WIP] Implement one kind of stre...
Posted by uncleGen <gi...@git.apache.org>.
Github user uncleGen commented on a diff in the pull request:
https://github.com/apache/spark/pull/17141#discussion_r104073813
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/streaming/ReservoirSampleSuit.scala ---
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.streaming
+
+import org.scalatest.BeforeAndAfterAll
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.streaming.InternalOutputModes._
+import org.apache.spark.sql.execution.streaming.MemoryStream
+import org.apache.spark.sql.execution.streaming.state.StateStore
+
+class ReservoirSampleSuit extends StateStoreMetricsTest with BeforeAndAfterAll {
+
+ import testImplicits._
+
+ override def afterAll(): Unit = {
+ super.afterAll()
+ StateStore.stop()
+ }
+
+ test("streaming reservoir sample: reservoir size is larger than stream data size - update mode") {
+ val inputData = MemoryStream[String]
+ val result = inputData.toDS().reservoir(4)
+
+ testStream(result, Update)(
+ AddData(inputData, "a", "b"),
+ CheckAnswer(Row("a"), Row("b")),
+ AddData(inputData, "a"),
+ CheckAnswer(Row("a"), Row("b"), Row("a"))
+ )
+ }
+
+ test("streaming reservoir sample: reservoir size is less than stream data size - update mode") {
+ val inputData = MemoryStream[String]
+ val result = inputData.toDS().reservoir(1)
+
+ testStream(result, Update)(
+ AddData(inputData, "a", "a"),
+ CheckLastBatch(Row("a")),
+ AddData(inputData, "b", "b", "b", "b", "b", "b", "b", "b"),
+ CheckLastBatch(Row("b"))
+ )
+ }
+
+ test("streaming reservoir sample with aggregation - update mode") {
+ val inputData = MemoryStream[String]
+ val result = inputData.toDS().reservoir(3).groupBy("value").count()
+
+ testStream(result, Update)(
+ AddData(inputData, "a"),
+ CheckAnswer(Row("a", 1)),
+ AddData(inputData, "b"),
+ CheckAnswer(Row("a", 1), Row("b", 1))
+ )
+ }
+
+ test("streaming reservoir sample with watermark") {
+ val inputData = MemoryStream[Int]
+ val result = inputData.toDS()
+ .withColumn("eventTime", $"value".cast("timestamp"))
+ .withWatermark("eventTime", "10 seconds")
+ .reservoir(10)
+ .select($"eventTime".cast("long").as[Long])
+
+ testStream(result, Update)(
+ AddData(inputData, (1 to 1).flatMap(_ => (11 to 15)): _*),
+ CheckLastBatch(11 to 15: _*),
+ AddData(inputData, 25), // Advance watermark to 15 seconds
+ CheckLastBatch(25),
+ AddData(inputData, 25), // Drop states less than watermark
+ CheckLastBatch(25),
+ AddData(inputData, 10), // Should not emit anything as data less than watermark
+ CheckLastBatch(),
+ AddData(inputData, 45), // Advance watermark to 35 seconds
+ CheckLastBatch(45),
+ AddData(inputData, 25), // Should not emit anything as data less than watermark
+ CheckLastBatch()
+ )
+ }
+
+ test("streaming reservoir sample with aggregation - complete mode") {
+ val inputData = MemoryStream[(String, Int)]
+ val result = inputData.toDS().select($"_1" as "key", $"_2" as "value")
+ .reservoir(3).groupBy("key").max("value")
+
+ testStream(result, Complete)(
+ AddData(inputData, ("a", 1)),
+ CheckAnswer(Row("a", 1)),
+ AddData(inputData, ("b", 2)),
+ CheckAnswer(Row("a", 1), Row("b", 2)),
+ StopStream,
+ StartStream(),
+ AddData(inputData, ("a", 10)),
+ CheckAnswer(Row("a", 10), Row("b", 2)),
+ AddData(inputData, (1 to 10).map(e => ("c", 100)): _*),
+ CheckAnswer(Row("a", 10), Row("b", 2), Row("c", 100))
+ )
+ }
+
+ test("batch reservoir sample") {
+ val df = spark.createDataset(Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 0))
+ assert(df.reservoir(3).count() == 3, "")
+ }
+
+ test("batch reservoir sample after aggregation") {
+ val df = spark.createDataset((1 to 10).map(e => (e, s"val_$e")))
+ .select($"_1" as "key", $"_2" as "value")
+ .groupBy("value").count()
+ assert(df.reservoir(3).count() == 3, "")
+ }
+
+ test("batch reservoir sample before aggregation") {
+ val df = spark.createDataset((1 to 10).map(e => (e, s"val_$e")))
+ .select($"_1" as "key", $"_2" as "value")
+ .reservoir(3)
+ .groupBy("value").count()
+ assert(df.count() == 3, "")
+ }
+}
--- End diff --
new unit test needs to be improved.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/17141
**[Test build #74843 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/74843/testReport)** for PR 17141 at commit [`02d44aa`](https://github.com/apache/spark/commit/02d44aa06f025dc1d69a7abbcf59691ce7ee0e4e).
* 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 `
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the issue:
https://github.com/apache/spark/pull/17141
Why does this need to be in Spark?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/17141
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/17141
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17141: [SPARK-19800][SS][WIP] Implement one kind of stre...
Posted by uncleGen <gi...@git.apache.org>.
Github user uncleGen commented on a diff in the pull request:
https://github.com/apache/spark/pull/17141#discussion_r104073607
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala ---
@@ -397,3 +402,110 @@ object StreamingDeduplicateExec {
private val EMPTY_ROW =
UnsafeProjection.create(Array[DataType](NullType)).apply(InternalRow.apply(null))
}
+
+/**
+ * Physical operator for executing streaming Sampling.
+ *
+ * @param k random sample k elements.
+ */
+case class StreamingReservoirSampleExec(
+ keyExpressions: Seq[Attribute],
+ child: SparkPlan,
+ k: Int,
+ stateId: Option[OperatorStateId] = None,
+ eventTimeWatermark: Option[Long] = None,
+ outputMode: Option[OutputMode] = None)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ override def requiredChildDistribution: Seq[Distribution] =
+ ClusteredDistribution(keyExpressions) :: Nil
+
+ private val enc = Encoders.STRING.asInstanceOf[ExpressionEncoder[String]]
+ private val NUM_RECORDS_IN_PARTITION = enc.toRow("NUM_RECORDS_IN_PARTITION")
+ .asInstanceOf[UnsafeRow]
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ metrics
+ val fieldTypes = (keyExpressions.map(_.dataType) ++ Seq(LongType)).toArray
+ val withSumFieldTypes = (keyExpressions.map(_.dataType) ++ Seq(LongType)).toArray
+
+ child.execute().mapPartitionsWithStateStore(
+ getStateId.checkpointLocation,
+ getStateId.operatorId,
+ getStateId.batchId,
+ keyExpressions.toStructType,
+ child.output.toStructType,
+ sqlContext.sessionState,
+ Some(sqlContext.streams.stateStoreCoordinator)) { (store, iter) =>
+
+ val numRecordsInPart = store.get(NUM_RECORDS_IN_PARTITION).map(value => {
+ value.get(0, LongType).asInstanceOf[Long]
+ }).getOrElse(0L)
+
+ val seed = Random.nextLong()
+ val rand = new XORShiftRandom(seed)
+ var numSamples = numRecordsInPart
+ var count = 0
+
+ val baseIterator = watermarkPredicate match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ baseIterator.foreach { r =>
+ count += 1
+ if (numSamples < k) {
+ numSamples += 1
+ store.put(enc.toRow(numSamples.toString).asInstanceOf[UnsafeRow],
+ r.asInstanceOf[UnsafeRow])
+ } else {
+ val randomIdx = (rand.nextDouble() * (numRecordsInPart + count)).toLong
+ if (randomIdx <= k) {
+ val replacementIdx = enc.toRow(randomIdx.toString).asInstanceOf[UnsafeRow]
+ store.put(replacementIdx, r.asInstanceOf[UnsafeRow])
+ }
+ }
+ }
+
+ val numRecordsTillNow = UnsafeProjection.create(Array[DataType](LongType))
+ .apply(InternalRow.apply(numRecordsInPart + count))
+ store.put(NUM_RECORDS_IN_PARTITION, numRecordsTillNow)
+ store.commit()
+
+ outputMode match {
+ case Some(Complete) =>
+ CompletionIterator[InternalRow, Iterator[InternalRow]](
+ store.iterator().filter(kv => {
+ !kv._1.asInstanceOf[UnsafeRow].equals(NUM_RECORDS_IN_PARTITION)
+ }).map(kv => {
+ UnsafeProjection.create(withSumFieldTypes).apply(InternalRow.fromSeq(
+ new JoinedRow(kv._2, numRecordsTillNow)
+ .toSeq(withSumFieldTypes)))
+ }), {})
+ case Some(Update) =>
+ CompletionIterator[InternalRow, Iterator[InternalRow]](
+ store.updates()
+ .filter(update => !update.key.equals(NUM_RECORDS_IN_PARTITION))
+ .map(update => {
+ UnsafeProjection.create(withSumFieldTypes).apply(InternalRow.fromSeq(
+ new JoinedRow(update.value, numRecordsTillNow)
+ .toSeq(withSumFieldTypes)))
+ }), {})
+ case _ =>
+ throw new UnsupportedOperationException(s"Invalid output mode: $outputMode " +
+ s"for streaming sampling.")
+ }
+ }.repartition(1).mapPartitions(it => {
+ SamplingUtils.reservoirSampleWithWeight(
+ it.map(item => (item, item.getLong(keyExpressions.size))), k)
+ .map(row =>
+ UnsafeProjection.create(fieldTypes)
+ .apply(InternalRow.fromSeq(row.toSeq(fieldTypes)))
+ ).iterator
+ })
+ }
--- End diff --
here, we do once global weight reservoir sampling.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/17141
**[Test build #74238 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/74238/testReport)** for PR 17141 at commit [`c4008cd`](https://github.com/apache/spark/commit/c4008cd466891c1be78ba6d5713250e78d7fa258).
* 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(`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/17141
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/74841/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by uncleGen <gi...@git.apache.org>.
Github user uncleGen commented on the issue:
https://github.com/apache/spark/pull/17141
ping @tdas and @zsxwing
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #17141: [SPARK-19800][SS][WIP] Implement one kind of stre...
Posted by uncleGen <gi...@git.apache.org>.
Github user uncleGen commented on a diff in the pull request:
https://github.com/apache/spark/pull/17141#discussion_r104073290
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala ---
@@ -397,3 +402,110 @@ object StreamingDeduplicateExec {
private val EMPTY_ROW =
UnsafeProjection.create(Array[DataType](NullType)).apply(InternalRow.apply(null))
}
+
+/**
+ * Physical operator for executing streaming Sampling.
+ *
+ * @param k random sample k elements.
+ */
+case class StreamingReservoirSampleExec(
+ keyExpressions: Seq[Attribute],
+ child: SparkPlan,
+ k: Int,
+ stateId: Option[OperatorStateId] = None,
+ eventTimeWatermark: Option[Long] = None,
+ outputMode: Option[OutputMode] = None)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ override def requiredChildDistribution: Seq[Distribution] =
+ ClusteredDistribution(keyExpressions) :: Nil
+
+ private val enc = Encoders.STRING.asInstanceOf[ExpressionEncoder[String]]
+ private val NUM_RECORDS_IN_PARTITION = enc.toRow("NUM_RECORDS_IN_PARTITION")
+ .asInstanceOf[UnsafeRow]
+
--- End diff --
`NUM_RECORDS_IN_PARTITION ` calculate the total number of records in current partiton, and update at the end of sample.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/17141
**[Test build #73778 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/73778/testReport)** for PR 17141 at commit [`288c124`](https://github.com/apache/spark/commit/288c124731901c0506833da05119fc131d1bf20e).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by uncleGen <gi...@git.apache.org>.
Github user uncleGen commented on the issue:
https://github.com/apache/spark/pull/17141
@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.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/17141
**[Test build #74841 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/74841/testReport)** for PR 17141 at commit [`1ddb82e`](https://github.com/apache/spark/commit/1ddb82e5a5aaadebd28080d334d58d082a125da8).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/17141
**[Test build #74238 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/74238/testReport)** for PR 17141 at commit [`c4008cd`](https://github.com/apache/spark/commit/c4008cd466891c1be78ba6d5713250e78d7fa258).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #17141: [SPARK-19800][SS][WIP] Implement one kind of streaming s...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/17141
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org