You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2021/06/25 08:30:09 UTC
[GitHub] [spark] HeartSaVioR opened a new pull request #33081: [SPARK-34893][SS] Support session window natively
HeartSaVioR opened a new pull request #33081:
URL: https://github.com/apache/spark/pull/33081
### What changes were proposed in this pull request?
This PR proposes to support native session window. Please refer the comments/design doc in SPARK-10816 for more details on the rationalization and design (could be outdated a bit compared to the PR).
The definition of the boundary of "session window" is [the timestamp of start event ~ the timestamp of last event + gap duration). That said, unlike time window, session window is a dynamic window which can expand if new input row is added to the session. To handle expansion of session window, Spark defines session window per input row, and "merge" windows if they can be merged (boundaries are overlapped).
This PR leverages two different approaches on merging session windows:
1. merging session windows with Spark's aggregation logic (a variant of sort aggregation)
2. updating session window for all rows bound to the same session, and applying aggregation logic afterwards
First one is preferable as it outperforms compared to the second one, though it can be only used if merging session window can be applied altogether with aggregation. It is not applicable on all the cases, so second one is used to cover the remaining cases.
This PR also applies the optimization on merging input rows and existing sessions with retaining the order (group keys + start timestamp of session window), leveraging the fact the number of existing sessions per group key won't be huge.
The state format is versioned, so that we can bring a new state format if we find a better one.
### Why are the changes needed?
For now, to deal with sessionization, Spark requires end users to play with (flat)MapGroupsWithState directly which has a couple of major drawbacks:
1. (flat)MapGroupsWithState is lower level API and end users have to code everything in details for defining session window and merging windows
2. built-in aggregate functions cannot be used and end users have to deal with aggregation by themselves
3. (flat)MapGroupsWithState is only available in Scala/Java.
With native support of session window, end users simply use "session_window" like they use "window" for tumbling/sliding window, and leverage built-in aggregate functions as well as UDAFs to simply define aggregations.
Quoting the query example from test suite:
```
val inputData = MemoryStream[(String, Long)]
// Split the lines into words, treat words as sessionId of events
val events = inputData.toDF()
.select($"_1".as("value"), $"_2".as("timestamp"))
.withColumn("eventTime", $"timestamp".cast("timestamp"))
.selectExpr("explode(split(value, ' ')) AS sessionId", "eventTime")
.withWatermark("eventTime", "30 seconds")
val sessionUpdates = events
.groupBy(session_window($"eventTime", "10 seconds") as 'session, 'sessionId)
.agg(count("*").as("numEvents"))
.selectExpr("sessionId", "CAST(session.start AS LONG)", "CAST(session.end AS LONG)",
"CAST(session.end AS LONG) - CAST(session.start AS LONG) AS durationMs",
"numEvents")
```
which is same as StructuredSessionization (native session window is shorter and clearer even ignoring model classes).
https://github.com/apache/spark/blob/39542bb81f8570219770bb6533c077f44f6cbd2a/examples/src/main/scala/org/apache/spark/examples/sql/streaming/StructuredSessionization.scala#L66-L105
(Worth noting that the code in StructuredSessionization only works with processing time. The code doesn't consider old event can update the start time of old session.)
### Does this PR introduce _any_ user-facing change?
Yes. This PR brings the new feature to support session window on both batch and streaming query, which adds a new function "session_window" which usage is similar with "window".
### How was this patch tested?
New test suites. Also tested with benchmark code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880364960
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45557/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670564349
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -511,6 +513,298 @@ case class StateStoreSaveExec(
copy(child = newChild)
}
+/**
+ * This class sorts input rows and existing sessions in state and provides output rows as
+ * sorted by "group keys + start time of session window".
+ *
+ * Refer [[MergingSortWithSessionWindowStateIterator]] for more details.
+ */
+case class SessionWindowStateStoreRestoreExec(
+ keyWithoutSessionExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ eventTimeWatermark: Option[Long],
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreReader with WatermarkSupport {
+
+ override def keyExpressions: Seq[Attribute] = keyWithoutSessionExpressions
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ assert(keyExpressions.nonEmpty, "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithReadStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ // We need to filter out outdated inputs
+ val filteredIterator = watermarkPredicateForData match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ new MergingSortWithSessionWindowStateIterator(
+ filteredIterator,
+ stateManager,
+ store,
+ keyWithoutSessionExpressions,
+ sessionExpression,
+ child.output).map { row =>
+ numOutputRows += 1
+ row
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def outputOrdering: Seq[SortOrder] = {
+ (keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending))
+ }
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyWithoutSessionExpressions.isEmpty) {
Review comment:
IIUC, the `keyWithoutSessionExpressions` is the grouping key except for the session window, it will never be empty since we don't support the single partition session window.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -511,6 +513,298 @@ case class StateStoreSaveExec(
copy(child = newChild)
}
+/**
+ * This class sorts input rows and existing sessions in state and provides output rows as
+ * sorted by "group keys + start time of session window".
+ *
+ * Refer [[MergingSortWithSessionWindowStateIterator]] for more details.
+ */
+case class SessionWindowStateStoreRestoreExec(
+ keyWithoutSessionExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ eventTimeWatermark: Option[Long],
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreReader with WatermarkSupport {
+
+ override def keyExpressions: Seq[Attribute] = keyWithoutSessionExpressions
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ assert(keyExpressions.nonEmpty, "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithReadStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ // We need to filter out outdated inputs
+ val filteredIterator = watermarkPredicateForData match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ new MergingSortWithSessionWindowStateIterator(
+ filteredIterator,
+ stateManager,
+ store,
+ keyWithoutSessionExpressions,
+ sessionExpression,
+ child.output).map { row =>
+ numOutputRows += 1
+ row
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def outputOrdering: Seq[SortOrder] = {
+ (keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending))
+ }
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyWithoutSessionExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyWithoutSessionExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def requiredChildOrdering: Seq[Seq[SortOrder]] = {
+ Seq((keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending)))
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
+ copy(child = newChild)
+}
+
+/**
+ * For each input tuple, the key is calculated and the tuple is `put` into the [[StateStore]].
+ */
+case class SessionWindowStateStoreSaveExec(
+ keyExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo] = None,
+ outputMode: Option[OutputMode] = None,
+ eventTimeWatermark: Option[Long] = None,
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ private val keyWithoutSessionExpressions = keyExpressions.filterNot { p =>
Review comment:
The `keyExpressions` passed-in should already filter out the session expression.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -511,6 +513,298 @@ case class StateStoreSaveExec(
copy(child = newChild)
}
+/**
+ * This class sorts input rows and existing sessions in state and provides output rows as
+ * sorted by "group keys + start time of session window".
+ *
+ * Refer [[MergingSortWithSessionWindowStateIterator]] for more details.
+ */
+case class SessionWindowStateStoreRestoreExec(
+ keyWithoutSessionExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ eventTimeWatermark: Option[Long],
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreReader with WatermarkSupport {
+
+ override def keyExpressions: Seq[Attribute] = keyWithoutSessionExpressions
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ assert(keyExpressions.nonEmpty, "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithReadStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ // We need to filter out outdated inputs
+ val filteredIterator = watermarkPredicateForData match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ new MergingSortWithSessionWindowStateIterator(
+ filteredIterator,
+ stateManager,
+ store,
+ keyWithoutSessionExpressions,
+ sessionExpression,
+ child.output).map { row =>
+ numOutputRows += 1
+ row
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def outputOrdering: Seq[SortOrder] = {
+ (keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending))
+ }
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyWithoutSessionExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyWithoutSessionExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def requiredChildOrdering: Seq[Seq[SortOrder]] = {
+ Seq((keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending)))
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
+ copy(child = newChild)
+}
+
+/**
+ * For each input tuple, the key is calculated and the tuple is `put` into the [[StateStore]].
+ */
+case class SessionWindowStateStoreSaveExec(
+ keyExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo] = None,
+ outputMode: Option[OutputMode] = None,
+ eventTimeWatermark: Option[Long] = None,
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ private val keyWithoutSessionExpressions = keyExpressions.filterNot { p =>
+ p.semanticEquals(sessionExpression)
+ }
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ metrics // force lazy init at driver
+ assert(outputMode.nonEmpty,
+ "Incorrect planning in IncrementalExecution, outputMode has not been set")
+ assert(keyExpressions.nonEmpty,
+ "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ val numOutputRows = longMetric("numOutputRows")
+ val allUpdatesTimeMs = longMetric("allUpdatesTimeMs")
+ val allRemovalsTimeMs = longMetric("allRemovalsTimeMs")
+ val commitTimeMs = longMetric("commitTimeMs")
+
+ outputMode match {
+ // Update and output all rows in the StateStore.
+ case Some(Complete) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, false)
+ }
+ allRemovalsTimeMs += 0
+ commitTimeMs += timeTakenMs {
+ stateManager.commit(store)
+ }
+ setStoreMetrics(store)
+ stateManager.iterator(store).map { row =>
+ numOutputRows += 1
+ row
+ }
+
+ // Update and output only rows being evicted from the StateStore
+ // Assumption: watermark predicates must be non-empty if append mode is allowed
+ case Some(Append) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, true)
+ }
+
+ val removalStartTimeNs = System.nanoTime
+ new NextIterator[InternalRow] {
+ private val removedIter = stateManager.removeByValueCondition(
+ store, watermarkPredicateForData.get.eval)
+
+ override protected def getNext(): InternalRow = {
+ if (!removedIter.hasNext) {
+ finished = true
+ null
+ } else {
+ numOutputRows += 1
+ removedIter.next()
+ }
+ }
+
+ override protected def close(): Unit = {
+ allRemovalsTimeMs += NANOSECONDS.toMillis(System.nanoTime - removalStartTimeNs)
+ commitTimeMs += timeTakenMs { store.commit() }
+ setStoreMetrics(store)
+ }
+ }
+
+ case Some(Update) =>
+ val iterPutToStore = iteratorPutToStore(iter, store, true, true)
+ new NextIterator[InternalRow] {
+ private val updatesStartTimeNs = System.nanoTime
+
+ override protected def getNext(): InternalRow = {
+ if (iterPutToStore.hasNext) {
+ iterPutToStore.next()
+ } else {
+ finished = true
+ null
+ }
+ }
+
+ override protected def close(): Unit = {
+ allUpdatesTimeMs += NANOSECONDS.toMillis(System.nanoTime - updatesStartTimeNs)
+
+ allRemovalsTimeMs += timeTakenMs {
+ if (watermarkPredicateForData.nonEmpty) {
+ val removedIter = stateManager.removeByValueCondition(
+ store, watermarkPredicateForData.get.eval)
+ while (removedIter.hasNext) {
+ removedIter.next()
+ }
+ }
+ }
+ commitTimeMs += timeTakenMs { store.commit() }
+ setStoreMetrics(store)
+ }
+ }
+
+ case _ => throw new UnsupportedOperationException(s"Invalid output mode: $outputMode")
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def shouldRunAnotherBatch(newMetadata: OffsetSeqMetadata): Boolean = {
Review comment:
`shouldRunAnotherBatch` is never used?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -511,6 +513,298 @@ case class StateStoreSaveExec(
copy(child = newChild)
}
+/**
+ * This class sorts input rows and existing sessions in state and provides output rows as
+ * sorted by "group keys + start time of session window".
+ *
+ * Refer [[MergingSortWithSessionWindowStateIterator]] for more details.
+ */
+case class SessionWindowStateStoreRestoreExec(
+ keyWithoutSessionExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ eventTimeWatermark: Option[Long],
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreReader with WatermarkSupport {
+
+ override def keyExpressions: Seq[Attribute] = keyWithoutSessionExpressions
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ assert(keyExpressions.nonEmpty, "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithReadStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ // We need to filter out outdated inputs
+ val filteredIterator = watermarkPredicateForData match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ new MergingSortWithSessionWindowStateIterator(
+ filteredIterator,
+ stateManager,
+ store,
+ keyWithoutSessionExpressions,
+ sessionExpression,
+ child.output).map { row =>
+ numOutputRows += 1
+ row
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def outputOrdering: Seq[SortOrder] = {
+ (keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending))
+ }
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyWithoutSessionExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyWithoutSessionExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def requiredChildOrdering: Seq[Seq[SortOrder]] = {
+ Seq((keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending)))
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
+ copy(child = newChild)
+}
+
+/**
+ * For each input tuple, the key is calculated and the tuple is `put` into the [[StateStore]].
+ */
+case class SessionWindowStateStoreSaveExec(
+ keyExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo] = None,
+ outputMode: Option[OutputMode] = None,
+ eventTimeWatermark: Option[Long] = None,
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ private val keyWithoutSessionExpressions = keyExpressions.filterNot { p =>
+ p.semanticEquals(sessionExpression)
+ }
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ metrics // force lazy init at driver
+ assert(outputMode.nonEmpty,
+ "Incorrect planning in IncrementalExecution, outputMode has not been set")
+ assert(keyExpressions.nonEmpty,
+ "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ val numOutputRows = longMetric("numOutputRows")
+ val allUpdatesTimeMs = longMetric("allUpdatesTimeMs")
+ val allRemovalsTimeMs = longMetric("allRemovalsTimeMs")
+ val commitTimeMs = longMetric("commitTimeMs")
+
+ outputMode match {
+ // Update and output all rows in the StateStore.
+ case Some(Complete) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, false)
+ }
+ allRemovalsTimeMs += 0
+ commitTimeMs += timeTakenMs {
+ stateManager.commit(store)
+ }
+ setStoreMetrics(store)
+ stateManager.iterator(store).map { row =>
+ numOutputRows += 1
+ row
+ }
+
+ // Update and output only rows being evicted from the StateStore
+ // Assumption: watermark predicates must be non-empty if append mode is allowed
+ case Some(Append) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, true)
+ }
+
+ val removalStartTimeNs = System.nanoTime
+ new NextIterator[InternalRow] {
+ private val removedIter = stateManager.removeByValueCondition(
+ store, watermarkPredicateForData.get.eval)
Review comment:
Seems we can simply count the time-spending of `removeByValueCondition` as removalStartTime? Then we can avoid calculating the time manually.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -511,6 +513,298 @@ case class StateStoreSaveExec(
copy(child = newChild)
}
+/**
+ * This class sorts input rows and existing sessions in state and provides output rows as
+ * sorted by "group keys + start time of session window".
+ *
+ * Refer [[MergingSortWithSessionWindowStateIterator]] for more details.
+ */
+case class SessionWindowStateStoreRestoreExec(
+ keyWithoutSessionExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ eventTimeWatermark: Option[Long],
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreReader with WatermarkSupport {
+
+ override def keyExpressions: Seq[Attribute] = keyWithoutSessionExpressions
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ assert(keyExpressions.nonEmpty, "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithReadStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ // We need to filter out outdated inputs
+ val filteredIterator = watermarkPredicateForData match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ new MergingSortWithSessionWindowStateIterator(
+ filteredIterator,
+ stateManager,
+ store,
+ keyWithoutSessionExpressions,
+ sessionExpression,
+ child.output).map { row =>
+ numOutputRows += 1
+ row
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def outputOrdering: Seq[SortOrder] = {
+ (keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending))
+ }
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyWithoutSessionExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyWithoutSessionExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def requiredChildOrdering: Seq[Seq[SortOrder]] = {
+ Seq((keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending)))
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
+ copy(child = newChild)
+}
+
+/**
+ * For each input tuple, the key is calculated and the tuple is `put` into the [[StateStore]].
+ */
+case class SessionWindowStateStoreSaveExec(
+ keyExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo] = None,
+ outputMode: Option[OutputMode] = None,
+ eventTimeWatermark: Option[Long] = None,
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ private val keyWithoutSessionExpressions = keyExpressions.filterNot { p =>
+ p.semanticEquals(sessionExpression)
+ }
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ metrics // force lazy init at driver
+ assert(outputMode.nonEmpty,
+ "Incorrect planning in IncrementalExecution, outputMode has not been set")
+ assert(keyExpressions.nonEmpty,
+ "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ val numOutputRows = longMetric("numOutputRows")
+ val allUpdatesTimeMs = longMetric("allUpdatesTimeMs")
+ val allRemovalsTimeMs = longMetric("allRemovalsTimeMs")
+ val commitTimeMs = longMetric("commitTimeMs")
+
+ outputMode match {
+ // Update and output all rows in the StateStore.
+ case Some(Complete) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, false)
+ }
+ allRemovalsTimeMs += 0
+ commitTimeMs += timeTakenMs {
+ stateManager.commit(store)
+ }
+ setStoreMetrics(store)
+ stateManager.iterator(store).map { row =>
+ numOutputRows += 1
+ row
+ }
+
+ // Update and output only rows being evicted from the StateStore
+ // Assumption: watermark predicates must be non-empty if append mode is allowed
+ case Some(Append) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, true)
+ }
+
+ val removalStartTimeNs = System.nanoTime
+ new NextIterator[InternalRow] {
+ private val removedIter = stateManager.removeByValueCondition(
+ store, watermarkPredicateForData.get.eval)
+
+ override protected def getNext(): InternalRow = {
+ if (!removedIter.hasNext) {
+ finished = true
+ null
+ } else {
+ numOutputRows += 1
+ removedIter.next()
+ }
+ }
+
+ override protected def close(): Unit = {
+ allRemovalsTimeMs += NANOSECONDS.toMillis(System.nanoTime - removalStartTimeNs)
+ commitTimeMs += timeTakenMs { store.commit() }
+ setStoreMetrics(store)
+ }
+ }
+
+ case Some(Update) =>
+ val iterPutToStore = iteratorPutToStore(iter, store, true, true)
+ new NextIterator[InternalRow] {
+ private val updatesStartTimeNs = System.nanoTime
+
+ override protected def getNext(): InternalRow = {
+ if (iterPutToStore.hasNext) {
+ iterPutToStore.next()
+ } else {
+ finished = true
+ null
+ }
+ }
+
+ override protected def close(): Unit = {
+ allUpdatesTimeMs += NANOSECONDS.toMillis(System.nanoTime - updatesStartTimeNs)
+
+ allRemovalsTimeMs += timeTakenMs {
+ if (watermarkPredicateForData.nonEmpty) {
+ val removedIter = stateManager.removeByValueCondition(
+ store, watermarkPredicateForData.get.eval)
+ while (removedIter.hasNext) {
+ removedIter.next()
+ }
+ }
+ }
+ commitTimeMs += timeTakenMs { store.commit() }
+ setStoreMetrics(store)
+ }
+ }
+
+ case _ => throw new UnsupportedOperationException(s"Invalid output mode: $outputMode")
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def shouldRunAnotherBatch(newMetadata: OffsetSeqMetadata): Boolean = {
+ (outputMode.contains(Append) || outputMode.contains(Update)) &&
+ eventTimeWatermark.isDefined &&
+ newMetadata.batchWatermarkMs > eventTimeWatermark.get
+ }
+
+ private def iteratorPutToStore(
+ baseIter: Iterator[InternalRow],
+ store: StateStore,
+ needFilter: Boolean,
Review comment:
nit: Instead of passing both `baseIter` and `needFilter`, seems we can directly pass in the iterator after filtering for both `iteratorPutToStore` and `putToStore`. WDYT?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -511,6 +513,298 @@ case class StateStoreSaveExec(
copy(child = newChild)
}
+/**
+ * This class sorts input rows and existing sessions in state and provides output rows as
+ * sorted by "group keys + start time of session window".
+ *
+ * Refer [[MergingSortWithSessionWindowStateIterator]] for more details.
+ */
+case class SessionWindowStateStoreRestoreExec(
+ keyWithoutSessionExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ eventTimeWatermark: Option[Long],
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreReader with WatermarkSupport {
+
+ override def keyExpressions: Seq[Attribute] = keyWithoutSessionExpressions
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ assert(keyExpressions.nonEmpty, "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithReadStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ // We need to filter out outdated inputs
+ val filteredIterator = watermarkPredicateForData match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ new MergingSortWithSessionWindowStateIterator(
+ filteredIterator,
+ stateManager,
+ store,
+ keyWithoutSessionExpressions,
+ sessionExpression,
+ child.output).map { row =>
+ numOutputRows += 1
Review comment:
nit: 2 more space for the indent?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -511,6 +513,298 @@ case class StateStoreSaveExec(
copy(child = newChild)
}
+/**
+ * This class sorts input rows and existing sessions in state and provides output rows as
+ * sorted by "group keys + start time of session window".
+ *
+ * Refer [[MergingSortWithSessionWindowStateIterator]] for more details.
+ */
+case class SessionWindowStateStoreRestoreExec(
+ keyWithoutSessionExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ eventTimeWatermark: Option[Long],
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreReader with WatermarkSupport {
+
+ override def keyExpressions: Seq[Attribute] = keyWithoutSessionExpressions
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ assert(keyExpressions.nonEmpty, "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithReadStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ // We need to filter out outdated inputs
+ val filteredIterator = watermarkPredicateForData match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ new MergingSortWithSessionWindowStateIterator(
+ filteredIterator,
+ stateManager,
+ store,
+ keyWithoutSessionExpressions,
+ sessionExpression,
+ child.output).map { row =>
+ numOutputRows += 1
+ row
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def outputOrdering: Seq[SortOrder] = {
+ (keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending))
+ }
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyWithoutSessionExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyWithoutSessionExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def requiredChildOrdering: Seq[Seq[SortOrder]] = {
+ Seq((keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending)))
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
+ copy(child = newChild)
+}
+
+/**
+ * For each input tuple, the key is calculated and the tuple is `put` into the [[StateStore]].
+ */
+case class SessionWindowStateStoreSaveExec(
+ keyExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo] = None,
+ outputMode: Option[OutputMode] = None,
+ eventTimeWatermark: Option[Long] = None,
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ private val keyWithoutSessionExpressions = keyExpressions.filterNot { p =>
+ p.semanticEquals(sessionExpression)
+ }
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ metrics // force lazy init at driver
+ assert(outputMode.nonEmpty,
+ "Incorrect planning in IncrementalExecution, outputMode has not been set")
+ assert(keyExpressions.nonEmpty,
+ "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ val numOutputRows = longMetric("numOutputRows")
+ val allUpdatesTimeMs = longMetric("allUpdatesTimeMs")
+ val allRemovalsTimeMs = longMetric("allRemovalsTimeMs")
+ val commitTimeMs = longMetric("commitTimeMs")
+
+ outputMode match {
+ // Update and output all rows in the StateStore.
+ case Some(Complete) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, false)
+ }
+ allRemovalsTimeMs += 0
Review comment:
+0 is for the metrics initialization?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670213455
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/UpdatingSessionsIterator.scala
##########
@@ -190,19 +191,21 @@ class UpdatingSessionsIterator(
}
private def closeCurrentSession(keyChanged: Boolean): Unit = {
- assert(returnRowsIter == null || !returnRowsIter.hasNext)
-
returnRows = rowsForCurrentSession
rowsForCurrentSession = null
- val groupingKey = generateGroupingKey()
+ val groupingKey = generateGroupingKey().copy()
val currentRowsIter = returnRows.generateIterator().map { internalRow =>
val valueRow = valueProj(internalRow)
restoreProj(join2(groupingKey, valueRow)).copy()
}
- returnRowsIter = currentRowsIter
+ if (returnRowsIter != null && returnRowsIter.hasNext) {
+ returnRowsIter = returnRowsIter ++ currentRowsIter
+ } else {
+ returnRowsIter = currentRowsIter
+ }
Review comment:
I found some edge-case that the last two input rows close session. If that happens, `returnRowsIter` is not null in this method (that's why I removed assertion) and we have to append both iterators instead of simply replacing.
That's why I have to add `.copy()` in groupingKey as evaluation takes lazily and somehow two iterators would use same key which it shouldn't.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880366635
**[Test build #141050 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141050/testReport)** for PR 33081 at commit [`1e78062`](https://github.com/apache/spark/commit/1e78062ea4f2051236ceff7d991288473ae821ce).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r671043429
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingSessionWindowSuite.scala
##########
@@ -0,0 +1,458 @@
+/*
+ * 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 java.util.Locale
+
+import org.scalatest.BeforeAndAfter
+import org.scalatest.matchers.must.Matchers
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.execution.streaming.MemoryStream
+import org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider
+import org.apache.spark.sql.functions.{count, session_window, sum}
+import org.apache.spark.sql.internal.SQLConf
+
+class StreamingSessionWindowSuite extends StreamTest
+ with BeforeAndAfter with Matchers with Logging {
+
+ import testImplicits._
+
+ after {
+ sqlContext.streams.active.foreach(_.stop())
+ }
+
+ def testWithAllOptions(name: String, confPairs: (String, String)*)
+ (func: => Any): Unit = {
+ val mergingSessionOptions = Seq(true, false).map { value =>
+ (SQLConf.STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION.key, value)
+ }
+ val providerOptions = Seq(
+ classOf[HDFSBackedStateStoreProvider].getCanonicalName).map { value =>
+ (SQLConf.STATE_STORE_PROVIDER_CLASS.key, value.stripSuffix("$"))
Review comment:
Don't we want to test against RocksDBStateStoreProvider?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879603643
**[Test build #140999 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140999/testReport)** for PR 33081 at commit [`0af2fd1`](https://github.com/apache/spark/commit/0af2fd12f2bcd09ea3b788b9c6ec99c022199398).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881538940
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141145/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670894977
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -511,6 +513,298 @@ case class StateStoreSaveExec(
copy(child = newChild)
}
+/**
+ * This class sorts input rows and existing sessions in state and provides output rows as
+ * sorted by "group keys + start time of session window".
+ *
+ * Refer [[MergingSortWithSessionWindowStateIterator]] for more details.
+ */
+case class SessionWindowStateStoreRestoreExec(
+ keyWithoutSessionExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ eventTimeWatermark: Option[Long],
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreReader with WatermarkSupport {
+
+ override def keyExpressions: Seq[Attribute] = keyWithoutSessionExpressions
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ assert(keyExpressions.nonEmpty, "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithReadStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ // We need to filter out outdated inputs
+ val filteredIterator = watermarkPredicateForData match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ new MergingSortWithSessionWindowStateIterator(
+ filteredIterator,
+ stateManager,
+ store,
+ keyWithoutSessionExpressions,
+ sessionExpression,
+ child.output).map { row =>
+ numOutputRows += 1
+ row
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def outputOrdering: Seq[SortOrder] = {
+ (keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending))
+ }
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyWithoutSessionExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyWithoutSessionExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def requiredChildOrdering: Seq[Seq[SortOrder]] = {
+ Seq((keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending)))
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
+ copy(child = newChild)
+}
+
+/**
+ * For each input tuple, the key is calculated and the tuple is `put` into the [[StateStore]].
+ */
+case class SessionWindowStateStoreSaveExec(
+ keyExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo] = None,
+ outputMode: Option[OutputMode] = None,
+ eventTimeWatermark: Option[Long] = None,
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ private val keyWithoutSessionExpressions = keyExpressions.filterNot { p =>
+ p.semanticEquals(sessionExpression)
+ }
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ metrics // force lazy init at driver
+ assert(outputMode.nonEmpty,
+ "Incorrect planning in IncrementalExecution, outputMode has not been set")
+ assert(keyExpressions.nonEmpty,
+ "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ val numOutputRows = longMetric("numOutputRows")
+ val allUpdatesTimeMs = longMetric("allUpdatesTimeMs")
+ val allRemovalsTimeMs = longMetric("allRemovalsTimeMs")
+ val commitTimeMs = longMetric("commitTimeMs")
+
+ outputMode match {
+ // Update and output all rows in the StateStore.
+ case Some(Complete) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, false)
+ }
+ allRemovalsTimeMs += 0
+ commitTimeMs += timeTakenMs {
+ stateManager.commit(store)
+ }
+ setStoreMetrics(store)
+ stateManager.iterator(store).map { row =>
+ numOutputRows += 1
+ row
+ }
+
+ // Update and output only rows being evicted from the StateStore
+ // Assumption: watermark predicates must be non-empty if append mode is allowed
+ case Some(Append) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, true)
+ }
+
+ val removalStartTimeNs = System.nanoTime
+ new NextIterator[InternalRow] {
+ private val removedIter = stateManager.removeByValueCondition(
+ store, watermarkPredicateForData.get.eval)
+
+ override protected def getNext(): InternalRow = {
+ if (!removedIter.hasNext) {
+ finished = true
+ null
+ } else {
+ numOutputRows += 1
+ removedIter.next()
+ }
+ }
+
+ override protected def close(): Unit = {
+ allRemovalsTimeMs += NANOSECONDS.toMillis(System.nanoTime - removalStartTimeNs)
+ commitTimeMs += timeTakenMs { store.commit() }
+ setStoreMetrics(store)
+ }
+ }
+
+ case Some(Update) =>
+ val iterPutToStore = iteratorPutToStore(iter, store, true, true)
+ new NextIterator[InternalRow] {
+ private val updatesStartTimeNs = System.nanoTime
+
+ override protected def getNext(): InternalRow = {
+ if (iterPutToStore.hasNext) {
+ iterPutToStore.next()
+ } else {
+ finished = true
+ null
+ }
+ }
+
+ override protected def close(): Unit = {
+ allUpdatesTimeMs += NANOSECONDS.toMillis(System.nanoTime - updatesStartTimeNs)
+
+ allRemovalsTimeMs += timeTakenMs {
+ if (watermarkPredicateForData.nonEmpty) {
+ val removedIter = stateManager.removeByValueCondition(
+ store, watermarkPredicateForData.get.eval)
+ while (removedIter.hasNext) {
+ removedIter.next()
+ }
+ }
+ }
+ commitTimeMs += timeTakenMs { store.commit() }
+ setStoreMetrics(store)
+ }
+ }
+
+ case _ => throw new UnsupportedOperationException(s"Invalid output mode: $outputMode")
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def shouldRunAnotherBatch(newMetadata: OffsetSeqMetadata): Boolean = {
Review comment:
It's for Spark to determine whether no-data batch is needed or not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880434824
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141042/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881142205
I will take another looks tonight.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-868372617
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44846/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670151257
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -3933,6 +3938,83 @@ object TimeWindowing extends Rule[LogicalPlan] {
}
}
+/** Maps a time column to a session window. */
+object SessionWindowing extends Rule[LogicalPlan] {
+ import org.apache.spark.sql.catalyst.dsl.expressions._
+
+ private final val SESSION_COL_NAME = "session_window"
+ private final val SESSION_START = "start"
+ private final val SESSION_END = "end"
+
+ /**
+ * Generates the logical plan for generating session window on a timestamp column.
+ * Each session window is initially defined as [timestamp, timestamp + gap).
+ *
+ * This also adds a marker to the session column so that downstream can easily find the column
+ * on session window.
+ */
+ def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp {
+ case p: LogicalPlan if p.children.size == 1 =>
+ val child = p.children.head
+ val sessionExpressions =
+ p.expressions.flatMap(_.collect { case s: SessionWindow => s }).toSet
+
+ val numWindowExpr = p.expressions.flatMap(_.collect {
+ case s: SessionWindow => s
+ case t: TimeWindow => t
+ }).toSet.size
+
+ // Only support a single session expression for now
+ if (numWindowExpr == 1 && sessionExpressions.nonEmpty &&
+ sessionExpressions.head.timeColumn.resolved &&
+ sessionExpressions.head.checkInputDataTypes().isSuccess) {
+
+ val session = sessionExpressions.head
+
+ val metadata = session.timeColumn match {
+ case a: Attribute => a.metadata
+ case _ => Metadata.empty
+ }
+
+ val newMetadata = new MetadataBuilder()
+ .withMetadata(metadata)
+ .putBoolean(SessionWindow.marker, true)
+ .build()
+
+ val sessionAttr = AttributeReference(
+ SESSION_COL_NAME, session.dataType, metadata = newMetadata)()
+
+ val sessionStart = PreciseTimestampConversion(session.timeColumn, TimestampType, LongType)
+ val sessionEnd = sessionStart + session.gapDuration
+
+ val literalSessionStruct = CreateNamedStruct(
+ Literal(SESSION_START) ::
+ PreciseTimestampConversion(sessionStart, LongType, TimestampType) ::
+ Literal(SESSION_END) ::
+ PreciseTimestampConversion(sessionEnd, LongType, TimestampType) ::
+ Nil)
+
+ val sessionStruct = Alias(literalSessionStruct, SESSION_COL_NAME)(
+ exprId = sessionAttr.exprId, explicitMetadata = Some(newMetadata))
+
+ val replacedPlan = p transformExpressions {
+ case s: SessionWindow => sessionAttr
+ }
+
+ // For backwards compatibility we add a filter to filter out nulls
Review comment:
I had to change the comment as we are going to do the same with time windows. Nice finding.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-868326810
**[Test build #140314 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140314/testReport)** for PR 33081 at commit [`af538b9`](https://github.com/apache/spark/commit/af538b968c3b4a3bb39045d09c9b5511e56f5020).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881138867
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45623/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880516228
**[Test build #141066 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141066/testReport)** for PR 33081 at commit [`9d331d0`](https://github.com/apache/spark/commit/9d331d0cb9c18bc8dab91f58a15c4a18a21e0745).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-868506998
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140314/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881323961
**[Test build #141145 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141145/testReport)** for PR 33081 at commit [`bbade35`](https://github.com/apache/spark/commit/bbade3501f16e9437ba9af4feca3e2029785d273).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881278084
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141131/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879626458
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45513/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880348707
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45557/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879832655
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45529/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879781606
**[Test build #141013 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141013/testReport)** for PR 33081 at commit [`11b44bb`](https://github.com/apache/spark/commit/11b44bb6344f5593dc58429a3c03a23b5a8d2aa7).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879826394
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45527/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879788487
**[Test build #141015 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141015/testReport)** for PR 33081 at commit [`7b74ee8`](https://github.com/apache/spark/commit/7b74ee8ccbb0b8edcdb5c4631138dca3dfe9778f).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670219318
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/functions.scala
##########
@@ -3630,6 +3630,35 @@ object functions {
window(timeColumn, windowDuration, windowDuration, "0 second")
}
+ /**
+ * Generates session window given a timestamp specifying column.
+ *
+ * Session window is the one of dynamic windows, which means the length of window is vary
+ * according to the given inputs. The length of session window is defined as "the timestamp
+ * of latest input of the session + gap duration", so when the new inputs are bound to the
+ * current session window, the end time of session window can be expanded according to the new
+ * inputs.
+ *
+ * Windows can support microsecond precision. Windows in the order of months are not supported.
Review comment:
We don't support the interval duration of window equal or more than a month. This is borrowed from `window` method doc, but I guess I need to change `Windows` to `gapDuration` to make clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-868506998
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140314/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880668699
**[Test build #141066 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141066/testReport)** for PR 33081 at commit [`9d331d0`](https://github.com/apache/spark/commit/9d331d0cb9c18bc8dab91f58a15c4a18a21e0745).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880669812
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141066/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-868323759
Same here; I marked this as draft as other PRs has to be reviewed and merged earlier. I'll rebase this PR once all other PRs are merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879944796
**[Test build #141022 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141022/testReport)** for PR 33081 at commit [`e01a366`](https://github.com/apache/spark/commit/e01a366b30d7fe03bcfa4a974b375565568739f3).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879663580
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45513/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-868487605
**[Test build #140314 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140314/testReport)** for PR 33081 at commit [`af538b9`](https://github.com/apache/spark/commit/af538b968c3b4a3bb39045d09c9b5511e56f5020).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `case class SessionWindow(timeColumn: Expression, gapDuration: Long) extends UnaryExpression`
* `case class SessionWindowStateStoreRestoreExec(`
* `case class SessionWindowStateStoreSaveExec(`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879663580
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45513/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880516228
**[Test build #141066 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141066/testReport)** for PR 33081 at commit [`9d331d0`](https://github.com/apache/spark/commit/9d331d0cb9c18bc8dab91f58a15c4a18a21e0745).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879951397
**[Test build #141022 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141022/testReport)** for PR 33081 at commit [`e01a366`](https://github.com/apache/spark/commit/e01a366b30d7fe03bcfa4a974b375565568739f3).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879781606
**[Test build #141013 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141013/testReport)** for PR 33081 at commit [`11b44bb`](https://github.com/apache/spark/commit/11b44bb6344f5593dc58429a3c03a23b5a8d2aa7).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881559356
Thank you @HeartSaVioR and @xuanyuanking!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879849670
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880430443
**[Test build #141042 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141042/testReport)** for PR 33081 at commit [`eef55e8`](https://github.com/apache/spark/commit/eef55e8284705f70f4dc9d08ed56fd941c98034f).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670157882
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -3933,6 +3938,83 @@ object TimeWindowing extends Rule[LogicalPlan] {
}
}
+/** Maps a time column to a session window. */
+object SessionWindowing extends Rule[LogicalPlan] {
+ import org.apache.spark.sql.catalyst.dsl.expressions._
+
+ private final val SESSION_COL_NAME = "session_window"
+ private final val SESSION_START = "start"
+ private final val SESSION_END = "end"
+
+ /**
+ * Generates the logical plan for generating session window on a timestamp column.
+ * Each session window is initially defined as [timestamp, timestamp + gap).
+ *
+ * This also adds a marker to the session column so that downstream can easily find the column
+ * on session window.
+ */
+ def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp {
+ case p: LogicalPlan if p.children.size == 1 =>
+ val child = p.children.head
+ val sessionExpressions =
+ p.expressions.flatMap(_.collect { case s: SessionWindow => s }).toSet
+
+ val numWindowExpr = p.expressions.flatMap(_.collect {
+ case s: SessionWindow => s
+ case t: TimeWindow => t
+ }).toSet.size
+
+ // Only support a single session expression for now
+ if (numWindowExpr == 1 && sessionExpressions.nonEmpty &&
+ sessionExpressions.head.timeColumn.resolved &&
+ sessionExpressions.head.checkInputDataTypes().isSuccess) {
+
+ val session = sessionExpressions.head
+
+ val metadata = session.timeColumn match {
+ case a: Attribute => a.metadata
+ case _ => Metadata.empty
+ }
+
+ val newMetadata = new MetadataBuilder()
+ .withMetadata(metadata)
+ .putBoolean(SessionWindow.marker, true)
+ .build()
+
+ val sessionAttr = AttributeReference(
+ SESSION_COL_NAME, session.dataType, metadata = newMetadata)()
+
+ val sessionStart = PreciseTimestampConversion(session.timeColumn, TimestampType, LongType)
+ val sessionEnd = sessionStart + session.gapDuration
+
+ val literalSessionStruct = CreateNamedStruct(
+ Literal(SESSION_START) ::
+ PreciseTimestampConversion(sessionStart, LongType, TimestampType) ::
+ Literal(SESSION_END) ::
+ PreciseTimestampConversion(sessionEnd, LongType, TimestampType) ::
+ Nil)
+
+ val sessionStruct = Alias(literalSessionStruct, SESSION_COL_NAME)(
+ exprId = sessionAttr.exprId, explicitMetadata = Some(newMetadata))
+
+ val replacedPlan = p transformExpressions {
+ case s: SessionWindow => sessionAttr
+ }
+
+ // For backwards compatibility we add a filter to filter out nulls
+ val filterExpr = IsNotNull(session.timeColumn)
+
+ replacedPlan.withNewChildren(
+ Filter(filterExpr,
+ Project(sessionStruct +: child.output, child)) :: Nil)
Review comment:
Actually this follows the logic on time windows. I agree we'd be better to filter first.
Shall we change the order to time windows as well? In different PR for sure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880336925
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45557/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670208288
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
##########
@@ -345,4 +351,179 @@ object AggUtils {
finalAndCompleteAggregate :: Nil
}
+
+ /**
+ * Plans a streaming session aggregation using the following progression:
+ *
+ * - Partial Aggregation
+ * - all tuples will have aggregated columns with initial value
+ * - (If "spark.sql.streaming.sessionWindow.merge.sessions.in.local.partition" is enabled)
+ * - Sort within partition (sort: all keys)
+ * - MergingSessionExec
+ * - calculate session among tuples, and aggregate tuples in session with partial merge
+ * - Shuffle & Sort (distribution: keys "without" session, sort: all keys)
+ * - SessionWindowStateStoreRestore (group: keys "without" session)
+ * - merge input tuples with stored tuples (sessions) respecting sort order
+ * - MergingSessionExec
+ * - calculate session among tuples, and aggregate tuples in session with partial merge
+ * - NOTE: it leverages the fact that the output of SessionWindowStateStoreRestore is sorted
+ * - now there is at most 1 tuple per group, key with session
+ * - SessionWindowStateStoreSave (group: keys "without" session)
+ * - saves tuple(s) for the next batch (multiple sessions could co-exist at the same time)
+ * - Complete (output the current result of the aggregation)
+ */
+ def planStreamingAggregationForSession(
+ groupingExpressions: Seq[NamedExpression],
+ sessionExpression: NamedExpression,
+ functionsWithoutDistinct: Seq[AggregateExpression],
+ resultExpressions: Seq[NamedExpression],
+ stateFormatVersion: Int,
+ mergeSessionsInLocalPartition: Boolean,
+ child: SparkPlan): Seq[SparkPlan] = {
+
+ val groupWithoutSessionExpression = groupingExpressions.filterNot { p =>
+ p.semanticEquals(sessionExpression)
+ }
+
+ if (groupWithoutSessionExpression.isEmpty) {
+ throw new AnalysisException("Global aggregation with session window in streaming query" +
+ " is not supported.")
+ }
+
+ val groupingWithoutSessionAttributes = groupWithoutSessionExpression.map(_.toAttribute)
+
+ val groupingAttributes = groupingExpressions.map(_.toAttribute)
+
+ // we don't do partial aggregate here, because it requires additional shuffle
Review comment:
Yeah I think this is wrong. This was written around 2 years ago I knew quite less on Spark. Nice finding.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881278084
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141131/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880403811
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45565/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879900595
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141015/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881138887
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45623/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880336424
cc. @viirya @xuanyuanking This is the last PR for session window feature. Please take a look. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879991855
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45537/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880541250
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45579/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879814960
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45530/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880320901
**[Test build #141042 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141042/testReport)** for PR 33081 at commit [`eef55e8`](https://github.com/apache/spark/commit/eef55e8284705f70f4dc9d08ed56fd941c98034f).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881145013
Rebased with master branch to see whether the pyspark lint is incurred from other PR and being fixed or not.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881383510
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45657/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881377004
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45657/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670865513
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -511,6 +513,298 @@ case class StateStoreSaveExec(
copy(child = newChild)
}
+/**
+ * This class sorts input rows and existing sessions in state and provides output rows as
+ * sorted by "group keys + start time of session window".
+ *
+ * Refer [[MergingSortWithSessionWindowStateIterator]] for more details.
+ */
+case class SessionWindowStateStoreRestoreExec(
+ keyWithoutSessionExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ eventTimeWatermark: Option[Long],
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreReader with WatermarkSupport {
+
+ override def keyExpressions: Seq[Attribute] = keyWithoutSessionExpressions
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ assert(keyExpressions.nonEmpty, "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithReadStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ // We need to filter out outdated inputs
+ val filteredIterator = watermarkPredicateForData match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ new MergingSortWithSessionWindowStateIterator(
+ filteredIterator,
+ stateManager,
+ store,
+ keyWithoutSessionExpressions,
+ sessionExpression,
+ child.output).map { row =>
+ numOutputRows += 1
+ row
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def outputOrdering: Seq[SortOrder] = {
+ (keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending))
+ }
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyWithoutSessionExpressions.isEmpty) {
Review comment:
Yeah. Probably better to move the assertion to initialization instead of doExecute, and change code to ignore the case key expression is empty. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881358273
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45657/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880700408
**[Test build #141080 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141080/testReport)** for PR 33081 at commit [`9d331d0`](https://github.com/apache/spark/commit/9d331d0cb9c18bc8dab91f58a15c4a18a21e0745).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881128374
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45623/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880386680
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45565/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880506611
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141050/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670139067
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SessionWindow.scala
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.catalyst.expressions
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.TypeCheckFailure
+import org.apache.spark.sql.catalyst.util.{DateTimeConstants, IntervalUtils}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+case class SessionWindow(timeColumn: Expression, gapDuration: Long) extends UnaryExpression
Review comment:
Could you add a few simple comments here? e.g, what `gapDuration` stands for.
##########
File path: python/pyspark/sql/functions.py
##########
@@ -2325,6 +2325,41 @@ def check_string_field(field, fieldName):
return Column(res)
+def session_window(timeColumn, gapDuration):
+ """
+ Generates session window given a timestamp specifying column.
+ Session window is the one of dynamic windows, which means the length of window is vary
Review comment:
is vary -> is varying?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
##########
@@ -345,4 +351,179 @@ object AggUtils {
finalAndCompleteAggregate :: Nil
}
+
+ /**
+ * Plans a streaming session aggregation using the following progression:
+ *
+ * - Partial Aggregation
+ * - all tuples will have aggregated columns with initial value
+ * - (If "spark.sql.streaming.sessionWindow.merge.sessions.in.local.partition" is enabled)
+ * - Sort within partition (sort: all keys)
+ * - MergingSessionExec
+ * - calculate session among tuples, and aggregate tuples in session with partial merge
+ * - Shuffle & Sort (distribution: keys "without" session, sort: all keys)
+ * - SessionWindowStateStoreRestore (group: keys "without" session)
+ * - merge input tuples with stored tuples (sessions) respecting sort order
+ * - MergingSessionExec
+ * - calculate session among tuples, and aggregate tuples in session with partial merge
+ * - NOTE: it leverages the fact that the output of SessionWindowStateStoreRestore is sorted
+ * - now there is at most 1 tuple per group, key with session
+ * - SessionWindowStateStoreSave (group: keys "without" session)
+ * - saves tuple(s) for the next batch (multiple sessions could co-exist at the same time)
+ * - Complete (output the current result of the aggregation)
+ */
+ def planStreamingAggregationForSession(
+ groupingExpressions: Seq[NamedExpression],
+ sessionExpression: NamedExpression,
+ functionsWithoutDistinct: Seq[AggregateExpression],
+ resultExpressions: Seq[NamedExpression],
+ stateFormatVersion: Int,
+ mergeSessionsInLocalPartition: Boolean,
+ child: SparkPlan): Seq[SparkPlan] = {
+
+ val groupWithoutSessionExpression = groupingExpressions.filterNot { p =>
+ p.semanticEquals(sessionExpression)
+ }
+
+ if (groupWithoutSessionExpression.isEmpty) {
+ throw new AnalysisException("Global aggregation with session window in streaming query" +
+ " is not supported.")
+ }
+
+ val groupingWithoutSessionAttributes = groupWithoutSessionExpression.map(_.toAttribute)
+
+ val groupingAttributes = groupingExpressions.map(_.toAttribute)
+
+ // we don't do partial aggregate here, because it requires additional shuffle
Review comment:
"we don't do partial aggregate here"? I think the following does partial aggregate, no?
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##########
@@ -1610,6 +1610,26 @@ object SQLConf {
.checkValue(v => Set(1, 2).contains(v), "Valid versions are 1 and 2")
.createWithDefault(2)
+ val STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION =
+ buildConf("spark.sql.streaming.sessionWindow.merge.sessions.in.local.partition")
Review comment:
Yea, is it necessary to have this config? Seems we can always do it.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
##########
@@ -113,6 +114,9 @@ object AggUtils {
resultExpressions = partialResultExpressions,
child = child)
+ val interExec: SparkPlan = mayAppendMergingSessionExec(groupingExpressions,
Review comment:
Could you add a comment here explaining why we need to append `MergingSessionExec`?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
##########
@@ -140,6 +144,8 @@ object AggUtils {
resultExpressions: Seq[NamedExpression],
child: SparkPlan): Seq[SparkPlan] = {
+ val maySessionChild = mayAppendUpdatingSessionExec(groupingExpressions, child)
Review comment:
ditto
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/UpdatingSessionsIterator.scala
##########
@@ -190,19 +191,21 @@ class UpdatingSessionsIterator(
}
private def closeCurrentSession(keyChanged: Boolean): Unit = {
- assert(returnRowsIter == null || !returnRowsIter.hasNext)
-
returnRows = rowsForCurrentSession
rowsForCurrentSession = null
- val groupingKey = generateGroupingKey()
+ val groupingKey = generateGroupingKey().copy()
val currentRowsIter = returnRows.generateIterator().map { internalRow =>
val valueRow = valueProj(internalRow)
restoreProj(join2(groupingKey, valueRow)).copy()
}
- returnRowsIter = currentRowsIter
+ if (returnRowsIter != null && returnRowsIter.hasNext) {
+ returnRowsIter = returnRowsIter ++ currentRowsIter
+ } else {
+ returnRowsIter = currentRowsIter
+ }
Review comment:
why adding this change?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/python/AggregateInPandasExec.scala
##########
@@ -107,13 +116,14 @@ case class AggregateInPandasExec(
// Map grouped rows to ArrowPythonRunner results, Only execute if partition is not empty
inputRDD.mapPartitionsInternal { iter => if (iter.isEmpty) iter else {
+ val newIter: Iterator[InternalRow] = mayAppendUpdatingSessionIterator(iter)
Review comment:
Could you also add a comment here?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/functions.scala
##########
@@ -3630,6 +3630,35 @@ object functions {
window(timeColumn, windowDuration, windowDuration, "0 second")
}
+ /**
+ * Generates session window given a timestamp specifying column.
+ *
+ * Session window is the one of dynamic windows, which means the length of window is vary
Review comment:
Session window is the one of dynamic windows -> Session window is one of dynamic windows
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -3933,6 +3938,83 @@ object TimeWindowing extends Rule[LogicalPlan] {
}
}
+/** Maps a time column to a session window. */
+object SessionWindowing extends Rule[LogicalPlan] {
+ import org.apache.spark.sql.catalyst.dsl.expressions._
+
+ private final val SESSION_COL_NAME = "session_window"
+ private final val SESSION_START = "start"
+ private final val SESSION_END = "end"
+
+ /**
+ * Generates the logical plan for generating session window on a timestamp column.
+ * Each session window is initially defined as [timestamp, timestamp + gap).
+ *
+ * This also adds a marker to the session column so that downstream can easily find the column
+ * on session window.
+ */
+ def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp {
+ case p: LogicalPlan if p.children.size == 1 =>
+ val child = p.children.head
+ val sessionExpressions =
+ p.expressions.flatMap(_.collect { case s: SessionWindow => s }).toSet
+
+ val numWindowExpr = p.expressions.flatMap(_.collect {
+ case s: SessionWindow => s
+ case t: TimeWindow => t
+ }).toSet.size
+
+ // Only support a single session expression for now
+ if (numWindowExpr == 1 && sessionExpressions.nonEmpty &&
+ sessionExpressions.head.timeColumn.resolved &&
+ sessionExpressions.head.checkInputDataTypes().isSuccess) {
+
+ val session = sessionExpressions.head
+
+ val metadata = session.timeColumn match {
+ case a: Attribute => a.metadata
+ case _ => Metadata.empty
+ }
+
+ val newMetadata = new MetadataBuilder()
+ .withMetadata(metadata)
+ .putBoolean(SessionWindow.marker, true)
+ .build()
+
+ val sessionAttr = AttributeReference(
+ SESSION_COL_NAME, session.dataType, metadata = newMetadata)()
+
+ val sessionStart = PreciseTimestampConversion(session.timeColumn, TimestampType, LongType)
+ val sessionEnd = sessionStart + session.gapDuration
+
+ val literalSessionStruct = CreateNamedStruct(
+ Literal(SESSION_START) ::
+ PreciseTimestampConversion(sessionStart, LongType, TimestampType) ::
+ Literal(SESSION_END) ::
+ PreciseTimestampConversion(sessionEnd, LongType, TimestampType) ::
+ Nil)
+
+ val sessionStruct = Alias(literalSessionStruct, SESSION_COL_NAME)(
+ exprId = sessionAttr.exprId, explicitMetadata = Some(newMetadata))
+
+ val replacedPlan = p transformExpressions {
+ case s: SessionWindow => sessionAttr
+ }
+
+ // For backwards compatibility we add a filter to filter out nulls
+ val filterExpr = IsNotNull(session.timeColumn)
+
+ replacedPlan.withNewChildren(
+ Filter(filterExpr,
+ Project(sessionStruct +: child.output, child)) :: Nil)
Review comment:
Hmm, why we filter out null time column after evaluating session? Shouldn't we filter before evaluating session?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/functions.scala
##########
@@ -3630,6 +3630,35 @@ object functions {
window(timeColumn, windowDuration, windowDuration, "0 second")
}
+ /**
+ * Generates session window given a timestamp specifying column.
+ *
+ * Session window is the one of dynamic windows, which means the length of window is vary
Review comment:
the length of window is vary -> the length of window is varying
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/functions.scala
##########
@@ -3630,6 +3630,35 @@ object functions {
window(timeColumn, windowDuration, windowDuration, "0 second")
}
+ /**
+ * Generates session window given a timestamp specifying column.
+ *
+ * Session window is the one of dynamic windows, which means the length of window is vary
+ * according to the given inputs. The length of session window is defined as "the timestamp
+ * of latest input of the session + gap duration", so when the new inputs are bound to the
+ * current session window, the end time of session window can be expanded according to the new
+ * inputs.
+ *
+ * Windows can support microsecond precision. Windows in the order of months are not supported.
Review comment:
> "Windows in the order of months are not supported."
I don't get what this sentence mean.
##########
File path: python/pyspark/sql/functions.py
##########
@@ -2325,6 +2325,41 @@ def check_string_field(field, fieldName):
return Column(res)
+def session_window(timeColumn, gapDuration):
+ """
+ Generates session window given a timestamp specifying column.
+ Session window is the one of dynamic windows, which means the length of window is vary
Review comment:
Session window is the one of dynamic windows -> Session window is one of dynamic windows
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/python/AggregateInPandasExec.scala
##########
@@ -53,12 +54,23 @@ case class AggregateInPandasExec(
override def producedAttributes: AttributeSet = AttributeSet(output)
- override def requiredChildDistribution: Seq[Distribution] = {
- if (groupingExpressions.isEmpty) {
- AllTuples :: Nil
- } else {
- ClusteredDistribution(groupingExpressions) :: Nil
- }
+ val sessionWindowOption = groupingExpressions.find { p =>
+ p.metadata.contains(SessionWindow.marker)
+ }
+
+ val groupingWithoutSessionExpressions = sessionWindowOption match {
+ case Some(sessionExpression) =>
+ groupingExpressions.filterNot { p => p.semanticEquals(sessionExpression) }
+
+ case None => groupingExpressions
+ }
+
+ override def requiredChildOrdering: Seq[Seq[SortOrder]] = sessionWindowOption match {
+ case Some(sessionExpression) =>
+ Seq((groupingWithoutSessionExpressions ++ Seq(sessionExpression))
+ .map(SortOrder(_, Ascending)))
+
+ case None => Seq(groupingExpressions.map(SortOrder(_, Ascending)))
Review comment:
Hmm, why we change original `requiredChildDistribution`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881220264
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141110/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881384829
Thanks all for reviewing! I merged this into master/3.2.
I filed https://issues.apache.org/jira/browse/SPARK-36172 to deal with documentation - probably need to explain about session window and how to use it in structured streaming guide doc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880546932
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45579/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670174122
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##########
@@ -1610,6 +1610,26 @@ object SQLConf {
.checkValue(v => Set(1, 2).contains(v), "Valid versions are 1 and 2")
.createWithDefault(2)
+ val STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION =
+ buildConf("spark.sql.streaming.sessionWindow.merge.sessions.in.local.partition")
Review comment:
As I explained on the doc method, this would incur additional "logical sort", so only useful when there're lots of input rows which are going to be consolidated into same session. The benefit is dependent on the characteristic of data.
If we want to pick one between two to simplify, it would be probably safer to remove local aggregation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879719914
**[Test build #140999 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140999/testReport)** for PR 33081 at commit [`0af2fd1`](https://github.com/apache/spark/commit/0af2fd12f2bcd09ea3b788b9c6ec99c022199398).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670895590
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -511,6 +513,298 @@ case class StateStoreSaveExec(
copy(child = newChild)
}
+/**
+ * This class sorts input rows and existing sessions in state and provides output rows as
+ * sorted by "group keys + start time of session window".
+ *
+ * Refer [[MergingSortWithSessionWindowStateIterator]] for more details.
+ */
+case class SessionWindowStateStoreRestoreExec(
+ keyWithoutSessionExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ eventTimeWatermark: Option[Long],
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreReader with WatermarkSupport {
+
+ override def keyExpressions: Seq[Attribute] = keyWithoutSessionExpressions
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ assert(keyExpressions.nonEmpty, "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithReadStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ // We need to filter out outdated inputs
+ val filteredIterator = watermarkPredicateForData match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ new MergingSortWithSessionWindowStateIterator(
+ filteredIterator,
+ stateManager,
+ store,
+ keyWithoutSessionExpressions,
+ sessionExpression,
+ child.output).map { row =>
+ numOutputRows += 1
+ row
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def outputOrdering: Seq[SortOrder] = {
+ (keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending))
+ }
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyWithoutSessionExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyWithoutSessionExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def requiredChildOrdering: Seq[Seq[SortOrder]] = {
+ Seq((keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending)))
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
+ copy(child = newChild)
+}
+
+/**
+ * For each input tuple, the key is calculated and the tuple is `put` into the [[StateStore]].
+ */
+case class SessionWindowStateStoreSaveExec(
+ keyExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo] = None,
+ outputMode: Option[OutputMode] = None,
+ eventTimeWatermark: Option[Long] = None,
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ private val keyWithoutSessionExpressions = keyExpressions.filterNot { p =>
+ p.semanticEquals(sessionExpression)
+ }
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ metrics // force lazy init at driver
+ assert(outputMode.nonEmpty,
+ "Incorrect planning in IncrementalExecution, outputMode has not been set")
+ assert(keyExpressions.nonEmpty,
+ "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ val numOutputRows = longMetric("numOutputRows")
+ val allUpdatesTimeMs = longMetric("allUpdatesTimeMs")
+ val allRemovalsTimeMs = longMetric("allRemovalsTimeMs")
+ val commitTimeMs = longMetric("commitTimeMs")
+
+ outputMode match {
+ // Update and output all rows in the StateStore.
+ case Some(Complete) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, false)
+ }
+ allRemovalsTimeMs += 0
+ commitTimeMs += timeTakenMs {
+ stateManager.commit(store)
+ }
+ setStoreMetrics(store)
+ stateManager.iterator(store).map { row =>
+ numOutputRows += 1
+ row
+ }
+
+ // Update and output only rows being evicted from the StateStore
+ // Assumption: watermark predicates must be non-empty if append mode is allowed
+ case Some(Append) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, true)
+ }
+
+ val removalStartTimeNs = System.nanoTime
+ new NextIterator[InternalRow] {
+ private val removedIter = stateManager.removeByValueCondition(
+ store, watermarkPredicateForData.get.eval)
+
+ override protected def getNext(): InternalRow = {
+ if (!removedIter.hasNext) {
+ finished = true
+ null
+ } else {
+ numOutputRows += 1
+ removedIter.next()
+ }
+ }
+
+ override protected def close(): Unit = {
+ allRemovalsTimeMs += NANOSECONDS.toMillis(System.nanoTime - removalStartTimeNs)
+ commitTimeMs += timeTakenMs { store.commit() }
+ setStoreMetrics(store)
+ }
+ }
+
+ case Some(Update) =>
+ val iterPutToStore = iteratorPutToStore(iter, store, true, true)
+ new NextIterator[InternalRow] {
+ private val updatesStartTimeNs = System.nanoTime
+
+ override protected def getNext(): InternalRow = {
+ if (iterPutToStore.hasNext) {
+ iterPutToStore.next()
+ } else {
+ finished = true
+ null
+ }
+ }
+
+ override protected def close(): Unit = {
+ allUpdatesTimeMs += NANOSECONDS.toMillis(System.nanoTime - updatesStartTimeNs)
+
+ allRemovalsTimeMs += timeTakenMs {
+ if (watermarkPredicateForData.nonEmpty) {
+ val removedIter = stateManager.removeByValueCondition(
+ store, watermarkPredicateForData.get.eval)
+ while (removedIter.hasNext) {
+ removedIter.next()
+ }
+ }
+ }
+ commitTimeMs += timeTakenMs { store.commit() }
+ setStoreMetrics(store)
+ }
+ }
+
+ case _ => throw new UnsupportedOperationException(s"Invalid output mode: $outputMode")
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def shouldRunAnotherBatch(newMetadata: OffsetSeqMetadata): Boolean = {
+ (outputMode.contains(Append) || outputMode.contains(Update)) &&
+ eventTimeWatermark.isDefined &&
+ newMetadata.batchWatermarkMs > eventTimeWatermark.get
+ }
+
+ private def iteratorPutToStore(
+ baseIter: Iterator[InternalRow],
+ store: StateStore,
+ needFilter: Boolean,
Review comment:
Yeah this was in old code and after refactoring this is no longer needed to be passed as parameter. Nice finding.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879849670
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881215299
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45642/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879944796
**[Test build #141022 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141022/testReport)** for PR 33081 at commit [`e01a366`](https://github.com/apache/spark/commit/e01a366b30d7fe03bcfa4a974b375565568739f3).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881215299
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45642/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879742694
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140999/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670116061
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -3933,6 +3938,83 @@ object TimeWindowing extends Rule[LogicalPlan] {
}
}
+/** Maps a time column to a session window. */
+object SessionWindowing extends Rule[LogicalPlan] {
+ import org.apache.spark.sql.catalyst.dsl.expressions._
+
+ private final val SESSION_COL_NAME = "session_window"
+ private final val SESSION_START = "start"
+ private final val SESSION_END = "end"
+
+ /**
+ * Generates the logical plan for generating session window on a timestamp column.
+ * Each session window is initially defined as [timestamp, timestamp + gap).
+ *
+ * This also adds a marker to the session column so that downstream can easily find the column
+ * on session window.
+ */
+ def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp {
+ case p: LogicalPlan if p.children.size == 1 =>
+ val child = p.children.head
+ val sessionExpressions =
+ p.expressions.flatMap(_.collect { case s: SessionWindow => s }).toSet
+
+ val numWindowExpr = p.expressions.flatMap(_.collect {
+ case s: SessionWindow => s
+ case t: TimeWindow => t
+ }).toSet.size
+
+ // Only support a single session expression for now
+ if (numWindowExpr == 1 && sessionExpressions.nonEmpty &&
+ sessionExpressions.head.timeColumn.resolved &&
+ sessionExpressions.head.checkInputDataTypes().isSuccess) {
+
+ val session = sessionExpressions.head
+
+ val metadata = session.timeColumn match {
+ case a: Attribute => a.metadata
+ case _ => Metadata.empty
+ }
+
+ val newMetadata = new MetadataBuilder()
+ .withMetadata(metadata)
+ .putBoolean(SessionWindow.marker, true)
+ .build()
+
+ val sessionAttr = AttributeReference(
+ SESSION_COL_NAME, session.dataType, metadata = newMetadata)()
+
+ val sessionStart = PreciseTimestampConversion(session.timeColumn, TimestampType, LongType)
+ val sessionEnd = sessionStart + session.gapDuration
+
+ val literalSessionStruct = CreateNamedStruct(
+ Literal(SESSION_START) ::
+ PreciseTimestampConversion(sessionStart, LongType, TimestampType) ::
+ Literal(SESSION_END) ::
+ PreciseTimestampConversion(sessionEnd, LongType, TimestampType) ::
+ Nil)
+
+ val sessionStruct = Alias(literalSessionStruct, SESSION_COL_NAME)(
+ exprId = sessionAttr.exprId, explicitMetadata = Some(newMetadata))
+
+ val replacedPlan = p transformExpressions {
+ case s: SessionWindow => sessionAttr
+ }
+
+ // For backwards compatibility we add a filter to filter out nulls
Review comment:
Didn't fully get the point of backward compatibility here, could you explain more?
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##########
@@ -1610,6 +1610,26 @@ object SQLConf {
.checkValue(v => Set(1, 2).contains(v), "Valid versions are 1 and 2")
.createWithDefault(2)
+ val STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION =
+ buildConf("spark.sql.streaming.sessionWindow.merge.sessions.in.local.partition")
Review comment:
How about `spark.sql.streaming.sessionWindow.localMerge.enabled` or `spark.sql.streaming.sessionWindow. mergeSessionsInLocalPartition.enabled`?
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##########
@@ -1610,6 +1610,26 @@ object SQLConf {
.checkValue(v => Set(1, 2).contains(v), "Valid versions are 1 and 2")
.createWithDefault(2)
+ val STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION =
+ buildConf("spark.sql.streaming.sessionWindow.merge.sessions.in.local.partition")
Review comment:
Comparing with the similar logic of AggUtils, maybe we can also remove this config? Just always do the local merge?
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##########
@@ -1610,6 +1610,26 @@ object SQLConf {
.checkValue(v => Set(1, 2).contains(v), "Valid versions are 1 and 2")
.createWithDefault(2)
+ val STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION =
+ buildConf("spark.sql.streaming.sessionWindow.merge.sessions.in.local.partition")
+ .internal()
+ .doc("When true, streaming session window sorts and merge sessions in local partition " +
+ "prior to shuffle. This is to reduce the rows to shuffle, but only beneficial when " +
+ "there're lots of rows in a batch being assigned to same sessions.")
+ .booleanConf
Review comment:
nit: version
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
##########
@@ -335,12 +339,29 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
}
}
- AggUtils.planStreamingAggregation(
- normalizedGroupingExpressions,
- aggregateExpressions.map(expr => expr.asInstanceOf[AggregateExpression]),
- rewrittenResultExpressions,
- stateVersion,
- planLater(child))
+ sessionWindowOption match {
+ case Some(sessionWindow) =>
+ val stateVersion = conf.getConf(SQLConf.STREAMING_SESSION_WINDOW_STATE_FORMAT_VERSION)
+
+ AggUtils.planStreamingAggregationForSession(
+ normalizedGroupingExpressions,
+ sessionWindow,
+ aggregateExpressions.map(expr => expr.asInstanceOf[AggregateExpression]),
+ rewrittenResultExpressions,
+ stateVersion,
+ conf.streamingSessionWindowMergeSessionInLocalPartition,
+ planLater(child))
+
+ case None =>
+ val stateVersion = conf.getConf(SQLConf.STREAMING_AGGREGATION_STATE_FORMAT_VERSION)
Review comment:
nit: we get the aggregation state format version twice https://github.com/apache/spark/pull/33081/files#diff-21f071d73070b8257ad76e6e16ec5ed38a13d1278fe94bd42546c258a69f4410R327
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SessionWindow.scala
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.catalyst.expressions
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.TypeCheckFailure
+import org.apache.spark.sql.catalyst.util.{DateTimeConstants, IntervalUtils}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+case class SessionWindow(timeColumn: Expression, gapDuration: Long) extends UnaryExpression
+ with ImplicitCastInputTypes
+ with Unevaluable
+ with NonSQLExpression {
+
+ //////////////////////////
+ // SQL Constructors
+ //////////////////////////
+
+ def this(timeColumn: Expression, gapDuration: Expression) = {
+ this(timeColumn, SessionWindow.parseExpression(gapDuration))
+ }
+
+ override def child: Expression = timeColumn
+ override def inputTypes: Seq[AbstractDataType] = Seq(TimestampType)
+ override def dataType: DataType = new StructType()
+ .add(StructField("start", TimestampType))
+ .add(StructField("end", TimestampType))
+
+ // This expression is replaced in the analyzer.
+ override lazy val resolved = false
+
+ /** Validate the inputs for the gap duration in addition to the input data type. */
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val dataTypeCheck = super.checkInputDataTypes()
+ if (dataTypeCheck.isSuccess) {
+ if (gapDuration <= 0) {
+ return TypeCheckFailure(s"The window duration ($gapDuration) must be greater than 0.")
+ }
+ }
+ dataTypeCheck
+ }
+
+ override protected def withNewChildInternal(newChild: Expression): Expression =
+ copy(timeColumn = newChild)
+}
+
+object SessionWindow {
+ val marker = "spark.sessionWindow"
+
+ /**
+ * Parses the interval string for a valid time duration. CalendarInterval expects interval
+ * strings to start with the string `interval`. For usability, we prepend `interval` to the string
+ * if the user omitted it.
+ *
+ * @param interval The interval string
+ * @return The interval duration in microseconds. SparkSQL casts TimestampType has microsecond
+ * precision.
+ */
+ private def getIntervalInMicroSeconds(interval: String): Long = {
Review comment:
Seems we can directly use TimeWindow. getIntervalInMicroSeconds?
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
##########
@@ -345,4 +351,179 @@ object AggUtils {
finalAndCompleteAggregate :: Nil
}
+
+ /**
+ * Plans a streaming session aggregation using the following progression:
+ *
+ * - Partial Aggregation
+ * - all tuples will have aggregated columns with initial value
Review comment:
nit: maybe we can mention `UpdateSessionsExec` here.
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SessionWindow.scala
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.catalyst.expressions
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.TypeCheckFailure
+import org.apache.spark.sql.catalyst.util.{DateTimeConstants, IntervalUtils}
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+case class SessionWindow(timeColumn: Expression, gapDuration: Long) extends UnaryExpression
+ with ImplicitCastInputTypes
+ with Unevaluable
+ with NonSQLExpression {
+
+ //////////////////////////
+ // SQL Constructors
+ //////////////////////////
+
+ def this(timeColumn: Expression, gapDuration: Expression) = {
+ this(timeColumn, SessionWindow.parseExpression(gapDuration))
+ }
+
+ override def child: Expression = timeColumn
+ override def inputTypes: Seq[AbstractDataType] = Seq(TimestampType)
+ override def dataType: DataType = new StructType()
+ .add(StructField("start", TimestampType))
+ .add(StructField("end", TimestampType))
+
+ // This expression is replaced in the analyzer.
+ override lazy val resolved = false
+
+ /** Validate the inputs for the gap duration in addition to the input data type. */
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val dataTypeCheck = super.checkInputDataTypes()
+ if (dataTypeCheck.isSuccess) {
+ if (gapDuration <= 0) {
+ return TypeCheckFailure(s"The window duration ($gapDuration) must be greater than 0.")
+ }
+ }
+ dataTypeCheck
+ }
+
+ override protected def withNewChildInternal(newChild: Expression): Expression =
+ copy(timeColumn = newChild)
+}
+
+object SessionWindow {
+ val marker = "spark.sessionWindow"
+
+ /**
+ * Parses the interval string for a valid time duration. CalendarInterval expects interval
+ * strings to start with the string `interval`. For usability, we prepend `interval` to the string
+ * if the user omitted it.
+ *
+ * @param interval The interval string
+ * @return The interval duration in microseconds. SparkSQL casts TimestampType has microsecond
+ * precision.
+ */
+ private def getIntervalInMicroSeconds(interval: String): Long = {
+ val cal = IntervalUtils.stringToInterval(UTF8String.fromString(interval))
+ if (cal.months != 0) {
+ throw new IllegalArgumentException(
+ s"Intervals greater than a month is not supported ($interval).")
+ }
+ cal.days * DateTimeConstants.MICROS_PER_DAY + cal.microseconds
+ }
+
+ /**
+ * Parses the duration expression to generate the long value for the original constructor so
+ * that we can use `window` in SQL.
+ */
+ private def parseExpression(expr: Expression): Long = expr match {
Review comment:
ditto, can we directly use TimeWindow.parseExpression?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881138887
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45623/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881153092
**[Test build #141131 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141131/testReport)** for PR 33081 at commit [`e7a2a37`](https://github.com/apache/spark/commit/e7a2a37227034c92daf404f556d1790dd6c29c3a).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879987577
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45537/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880669812
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141066/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879942790
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141013/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670894600
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -511,6 +513,298 @@ case class StateStoreSaveExec(
copy(child = newChild)
}
+/**
+ * This class sorts input rows and existing sessions in state and provides output rows as
+ * sorted by "group keys + start time of session window".
+ *
+ * Refer [[MergingSortWithSessionWindowStateIterator]] for more details.
+ */
+case class SessionWindowStateStoreRestoreExec(
+ keyWithoutSessionExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ eventTimeWatermark: Option[Long],
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreReader with WatermarkSupport {
+
+ override def keyExpressions: Seq[Attribute] = keyWithoutSessionExpressions
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ assert(keyExpressions.nonEmpty, "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithReadStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ // We need to filter out outdated inputs
+ val filteredIterator = watermarkPredicateForData match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ new MergingSortWithSessionWindowStateIterator(
+ filteredIterator,
+ stateManager,
+ store,
+ keyWithoutSessionExpressions,
+ sessionExpression,
+ child.output).map { row =>
+ numOutputRows += 1
+ row
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def outputOrdering: Seq[SortOrder] = {
+ (keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending))
+ }
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyWithoutSessionExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyWithoutSessionExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def requiredChildOrdering: Seq[Seq[SortOrder]] = {
+ Seq((keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending)))
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
+ copy(child = newChild)
+}
+
+/**
+ * For each input tuple, the key is calculated and the tuple is `put` into the [[StateStore]].
+ */
+case class SessionWindowStateStoreSaveExec(
+ keyExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo] = None,
+ outputMode: Option[OutputMode] = None,
+ eventTimeWatermark: Option[Long] = None,
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ private val keyWithoutSessionExpressions = keyExpressions.filterNot { p =>
+ p.semanticEquals(sessionExpression)
+ }
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ metrics // force lazy init at driver
+ assert(outputMode.nonEmpty,
+ "Incorrect planning in IncrementalExecution, outputMode has not been set")
+ assert(keyExpressions.nonEmpty,
+ "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ val numOutputRows = longMetric("numOutputRows")
+ val allUpdatesTimeMs = longMetric("allUpdatesTimeMs")
+ val allRemovalsTimeMs = longMetric("allRemovalsTimeMs")
+ val commitTimeMs = longMetric("commitTimeMs")
+
+ outputMode match {
+ // Update and output all rows in the StateStore.
+ case Some(Complete) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, false)
+ }
+ allRemovalsTimeMs += 0
+ commitTimeMs += timeTakenMs {
+ stateManager.commit(store)
+ }
+ setStoreMetrics(store)
+ stateManager.iterator(store).map { row =>
+ numOutputRows += 1
+ row
+ }
+
+ // Update and output only rows being evicted from the StateStore
+ // Assumption: watermark predicates must be non-empty if append mode is allowed
+ case Some(Append) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, true)
+ }
+
+ val removalStartTimeNs = System.nanoTime
+ new NextIterator[InternalRow] {
+ private val removedIter = stateManager.removeByValueCondition(
+ store, watermarkPredicateForData.get.eval)
Review comment:
removeByValueCondition returns "iterator" which is evaluated "lazily".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879808697
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45527/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670219318
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/functions.scala
##########
@@ -3630,6 +3630,35 @@ object functions {
window(timeColumn, windowDuration, windowDuration, "0 second")
}
+ /**
+ * Generates session window given a timestamp specifying column.
+ *
+ * Session window is the one of dynamic windows, which means the length of window is vary
+ * according to the given inputs. The length of session window is defined as "the timestamp
+ * of latest input of the session + gap duration", so when the new inputs are bound to the
+ * current session window, the end time of session window can be expanded according to the new
+ * inputs.
+ *
+ * Windows can support microsecond precision. Windows in the order of months are not supported.
Review comment:
We don't support the interval duration of window equal or more than a month. This is borrowed from `window` method doc, but I guess I need to change Windows to gapDuration to make clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-868326810
**[Test build #140314 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140314/testReport)** for PR 33081 at commit [`af538b9`](https://github.com/apache/spark/commit/af538b968c3b4a3bb39045d09c9b5511e56f5020).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880366635
**[Test build #141050 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141050/testReport)** for PR 33081 at commit [`1e78062`](https://github.com/apache/spark/commit/1e78062ea4f2051236ceff7d991288473ae821ce).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880364960
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45557/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879951459
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141022/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879814729
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45529/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670893597
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -511,6 +513,298 @@ case class StateStoreSaveExec(
copy(child = newChild)
}
+/**
+ * This class sorts input rows and existing sessions in state and provides output rows as
+ * sorted by "group keys + start time of session window".
+ *
+ * Refer [[MergingSortWithSessionWindowStateIterator]] for more details.
+ */
+case class SessionWindowStateStoreRestoreExec(
+ keyWithoutSessionExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ eventTimeWatermark: Option[Long],
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreReader with WatermarkSupport {
+
+ override def keyExpressions: Seq[Attribute] = keyWithoutSessionExpressions
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ assert(keyExpressions.nonEmpty, "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithReadStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ // We need to filter out outdated inputs
+ val filteredIterator = watermarkPredicateForData match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ new MergingSortWithSessionWindowStateIterator(
+ filteredIterator,
+ stateManager,
+ store,
+ keyWithoutSessionExpressions,
+ sessionExpression,
+ child.output).map { row =>
+ numOutputRows += 1
+ row
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def outputOrdering: Seq[SortOrder] = {
+ (keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending))
+ }
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyWithoutSessionExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyWithoutSessionExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def requiredChildOrdering: Seq[Seq[SortOrder]] = {
+ Seq((keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending)))
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
+ copy(child = newChild)
+}
+
+/**
+ * For each input tuple, the key is calculated and the tuple is `put` into the [[StateStore]].
+ */
+case class SessionWindowStateStoreSaveExec(
+ keyExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo] = None,
+ outputMode: Option[OutputMode] = None,
+ eventTimeWatermark: Option[Long] = None,
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ private val keyWithoutSessionExpressions = keyExpressions.filterNot { p =>
Review comment:
Actually the name has been confusing; let's do the same with `SessionWindowStateStoreRestoreExec`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670894308
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala
##########
@@ -511,6 +513,298 @@ case class StateStoreSaveExec(
copy(child = newChild)
}
+/**
+ * This class sorts input rows and existing sessions in state and provides output rows as
+ * sorted by "group keys + start time of session window".
+ *
+ * Refer [[MergingSortWithSessionWindowStateIterator]] for more details.
+ */
+case class SessionWindowStateStoreRestoreExec(
+ keyWithoutSessionExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ eventTimeWatermark: Option[Long],
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreReader with WatermarkSupport {
+
+ override def keyExpressions: Seq[Attribute] = keyWithoutSessionExpressions
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ val numOutputRows = longMetric("numOutputRows")
+ assert(keyExpressions.nonEmpty, "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithReadStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ // We need to filter out outdated inputs
+ val filteredIterator = watermarkPredicateForData match {
+ case Some(predicate) => iter.filter((row: InternalRow) => !predicate.eval(row))
+ case None => iter
+ }
+
+ new MergingSortWithSessionWindowStateIterator(
+ filteredIterator,
+ stateManager,
+ store,
+ keyWithoutSessionExpressions,
+ sessionExpression,
+ child.output).map { row =>
+ numOutputRows += 1
+ row
+ }
+ }
+ }
+
+ override def output: Seq[Attribute] = child.output
+
+ override def outputPartitioning: Partitioning = child.outputPartitioning
+
+ override def outputOrdering: Seq[SortOrder] = {
+ (keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending))
+ }
+
+ override def requiredChildDistribution: Seq[Distribution] = {
+ if (keyWithoutSessionExpressions.isEmpty) {
+ AllTuples :: Nil
+ } else {
+ ClusteredDistribution(keyWithoutSessionExpressions, stateInfo.map(_.numPartitions)) :: Nil
+ }
+ }
+
+ override def requiredChildOrdering: Seq[Seq[SortOrder]] = {
+ Seq((keyWithoutSessionExpressions ++ Seq(sessionExpression)).map(SortOrder(_, Ascending)))
+ }
+
+ override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan =
+ copy(child = newChild)
+}
+
+/**
+ * For each input tuple, the key is calculated and the tuple is `put` into the [[StateStore]].
+ */
+case class SessionWindowStateStoreSaveExec(
+ keyExpressions: Seq[Attribute],
+ sessionExpression: Attribute,
+ stateInfo: Option[StatefulOperatorStateInfo] = None,
+ outputMode: Option[OutputMode] = None,
+ eventTimeWatermark: Option[Long] = None,
+ stateFormatVersion: Int,
+ child: SparkPlan)
+ extends UnaryExecNode with StateStoreWriter with WatermarkSupport {
+
+ private val keyWithoutSessionExpressions = keyExpressions.filterNot { p =>
+ p.semanticEquals(sessionExpression)
+ }
+
+ private val stateManager = StreamingSessionWindowStateManager.createStateManager(
+ keyWithoutSessionExpressions, sessionExpression, child.output, stateFormatVersion)
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ metrics // force lazy init at driver
+ assert(outputMode.nonEmpty,
+ "Incorrect planning in IncrementalExecution, outputMode has not been set")
+ assert(keyExpressions.nonEmpty,
+ "Grouping key must be specified when using sessionWindow")
+
+ child.execute().mapPartitionsWithStateStore(
+ getStateInfo,
+ stateManager.getStateKeySchema,
+ stateManager.getStateValueSchema,
+ numColsPrefixKey = stateManager.getNumColsForPrefixKey,
+ session.sessionState,
+ Some(session.streams.stateStoreCoordinator)) { case (store, iter) =>
+
+ val numOutputRows = longMetric("numOutputRows")
+ val allUpdatesTimeMs = longMetric("allUpdatesTimeMs")
+ val allRemovalsTimeMs = longMetric("allRemovalsTimeMs")
+ val commitTimeMs = longMetric("commitTimeMs")
+
+ outputMode match {
+ // Update and output all rows in the StateStore.
+ case Some(Complete) =>
+ allUpdatesTimeMs += timeTakenMs {
+ putToStore(iter, store, false)
+ }
+ allRemovalsTimeMs += 0
Review comment:
This is just to make sure we make consistent with StateStoreSaveExec, but looks like not needed. Let's remove and see.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r671105748
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingSessionWindowSuite.scala
##########
@@ -0,0 +1,458 @@
+/*
+ * 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 java.util.Locale
+
+import org.scalatest.BeforeAndAfter
+import org.scalatest.matchers.must.Matchers
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.execution.streaming.MemoryStream
+import org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider
+import org.apache.spark.sql.functions.{count, session_window, sum}
+import org.apache.spark.sql.internal.SQLConf
+
+class StreamingSessionWindowSuite extends StreamTest
+ with BeforeAndAfter with Matchers with Logging {
+
+ import testImplicits._
+
+ after {
+ sqlContext.streams.active.foreach(_.stop())
+ }
+
+ def testWithAllOptions(name: String, confPairs: (String, String)*)
+ (func: => Any): Unit = {
+ val mergingSessionOptions = Seq(true, false).map { value =>
+ (SQLConf.STREAMING_SESSION_WINDOW_MERGE_SESSIONS_IN_LOCAL_PARTITION.key, value)
+ }
+ val providerOptions = Seq(
+ classOf[HDFSBackedStateStoreProvider].getCanonicalName).map { value =>
+ (SQLConf.STATE_STORE_PROVIDER_CLASS.key, value.stripSuffix("$"))
Review comment:
Missing spot. Nice finding!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670190914
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -3933,6 +3938,83 @@ object TimeWindowing extends Rule[LogicalPlan] {
}
}
+/** Maps a time column to a session window. */
+object SessionWindowing extends Rule[LogicalPlan] {
+ import org.apache.spark.sql.catalyst.dsl.expressions._
+
+ private final val SESSION_COL_NAME = "session_window"
+ private final val SESSION_START = "start"
+ private final val SESSION_END = "end"
+
+ /**
+ * Generates the logical plan for generating session window on a timestamp column.
+ * Each session window is initially defined as [timestamp, timestamp + gap).
+ *
+ * This also adds a marker to the session column so that downstream can easily find the column
+ * on session window.
+ */
+ def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp {
+ case p: LogicalPlan if p.children.size == 1 =>
+ val child = p.children.head
+ val sessionExpressions =
+ p.expressions.flatMap(_.collect { case s: SessionWindow => s }).toSet
+
+ val numWindowExpr = p.expressions.flatMap(_.collect {
+ case s: SessionWindow => s
+ case t: TimeWindow => t
+ }).toSet.size
+
+ // Only support a single session expression for now
+ if (numWindowExpr == 1 && sessionExpressions.nonEmpty &&
+ sessionExpressions.head.timeColumn.resolved &&
+ sessionExpressions.head.checkInputDataTypes().isSuccess) {
+
+ val session = sessionExpressions.head
+
+ val metadata = session.timeColumn match {
+ case a: Attribute => a.metadata
+ case _ => Metadata.empty
+ }
+
+ val newMetadata = new MetadataBuilder()
+ .withMetadata(metadata)
+ .putBoolean(SessionWindow.marker, true)
+ .build()
+
+ val sessionAttr = AttributeReference(
+ SESSION_COL_NAME, session.dataType, metadata = newMetadata)()
+
+ val sessionStart = PreciseTimestampConversion(session.timeColumn, TimestampType, LongType)
+ val sessionEnd = sessionStart + session.gapDuration
+
+ val literalSessionStruct = CreateNamedStruct(
+ Literal(SESSION_START) ::
+ PreciseTimestampConversion(sessionStart, LongType, TimestampType) ::
+ Literal(SESSION_END) ::
+ PreciseTimestampConversion(sessionEnd, LongType, TimestampType) ::
+ Nil)
+
+ val sessionStruct = Alias(literalSessionStruct, SESSION_COL_NAME)(
+ exprId = sessionAttr.exprId, explicitMetadata = Some(newMetadata))
+
+ val replacedPlan = p transformExpressions {
+ case s: SessionWindow => sessionAttr
+ }
+
+ // For backwards compatibility we add a filter to filter out nulls
+ val filterExpr = IsNotNull(session.timeColumn)
+
+ replacedPlan.withNewChildren(
+ Filter(filterExpr,
+ Project(sessionStruct +: child.output, child)) :: Nil)
Review comment:
SGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880546932
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45579/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881323961
**[Test build #141145 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141145/testReport)** for PR 33081 at commit [`bbade35`](https://github.com/apache/spark/commit/bbade3501f16e9437ba9af4feca3e2029785d273).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881513218
**[Test build #141145 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141145/testReport)** for PR 33081 at commit [`bbade35`](https://github.com/apache/spark/commit/bbade3501f16e9437ba9af4feca3e2029785d273).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880408617
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45565/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879603643
**[Test build #140999 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140999/testReport)** for PR 33081 at commit [`0af2fd1`](https://github.com/apache/spark/commit/0af2fd12f2bcd09ea3b788b9c6ec99c022199398).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880737875
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45594/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879922339
**[Test build #141013 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141013/testReport)** for PR 33081 at commit [`11b44bb`](https://github.com/apache/spark/commit/11b44bb6344f5593dc58429a3c03a23b5a8d2aa7).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879991855
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45537/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881274889
**[Test build #141131 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141131/testReport)** for PR 33081 at commit [`e7a2a37`](https://github.com/apache/spark/commit/e7a2a37227034c92daf404f556d1790dd6c29c3a).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880796962
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45594/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881220264
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141110/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-868371755
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44846/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670205908
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
##########
@@ -345,4 +351,179 @@ object AggUtils {
finalAndCompleteAggregate :: Nil
}
+
+ /**
+ * Plans a streaming session aggregation using the following progression:
+ *
+ * - Partial Aggregation
+ * - all tuples will have aggregated columns with initial value
Review comment:
This method does not leverage UpdateSessionsExec.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880506611
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141050/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR closed pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR closed pull request #33081:
URL: https://github.com/apache/spark/pull/33081
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881153092
**[Test build #141131 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141131/testReport)** for PR 33081 at commit [`e7a2a37`](https://github.com/apache/spark/commit/e7a2a37227034c92daf404f556d1790dd6c29c3a).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880408617
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45565/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880683255
retest this, please
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879951459
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141022/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881114900
**[Test build #141110 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141110/testReport)** for PR 33081 at commit [`4a6cff3`](https://github.com/apache/spark/commit/4a6cff3e55d91cb19bd8f512950c6ef7fb02504c).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r671040093
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/UpdatingSessionsIterator.scala
##########
@@ -190,19 +191,21 @@ class UpdatingSessionsIterator(
}
private def closeCurrentSession(keyChanged: Boolean): Unit = {
- assert(returnRowsIter == null || !returnRowsIter.hasNext)
-
returnRows = rowsForCurrentSession
rowsForCurrentSession = null
- val groupingKey = generateGroupingKey()
+ val groupingKey = generateGroupingKey().copy()
val currentRowsIter = returnRows.generateIterator().map { internalRow =>
val valueRow = valueProj(internalRow)
restoreProj(join2(groupingKey, valueRow)).copy()
}
- returnRowsIter = currentRowsIter
+ if (returnRowsIter != null && returnRowsIter.hasNext) {
+ returnRowsIter = returnRowsIter ++ currentRowsIter
+ } else {
+ returnRowsIter = currentRowsIter
+ }
Review comment:
Have you added corresponding test for the edge cases?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881383510
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45657/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879742694
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140999/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670383324
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
##########
@@ -3933,6 +3938,83 @@ object TimeWindowing extends Rule[LogicalPlan] {
}
}
+/** Maps a time column to a session window. */
+object SessionWindowing extends Rule[LogicalPlan] {
+ import org.apache.spark.sql.catalyst.dsl.expressions._
+
+ private final val SESSION_COL_NAME = "session_window"
+ private final val SESSION_START = "start"
+ private final val SESSION_END = "end"
+
+ /**
+ * Generates the logical plan for generating session window on a timestamp column.
+ * Each session window is initially defined as [timestamp, timestamp + gap).
+ *
+ * This also adds a marker to the session column so that downstream can easily find the column
+ * on session window.
+ */
+ def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp {
+ case p: LogicalPlan if p.children.size == 1 =>
+ val child = p.children.head
+ val sessionExpressions =
+ p.expressions.flatMap(_.collect { case s: SessionWindow => s }).toSet
+
+ val numWindowExpr = p.expressions.flatMap(_.collect {
+ case s: SessionWindow => s
+ case t: TimeWindow => t
+ }).toSet.size
+
+ // Only support a single session expression for now
+ if (numWindowExpr == 1 && sessionExpressions.nonEmpty &&
+ sessionExpressions.head.timeColumn.resolved &&
+ sessionExpressions.head.checkInputDataTypes().isSuccess) {
+
+ val session = sessionExpressions.head
+
+ val metadata = session.timeColumn match {
+ case a: Attribute => a.metadata
+ case _ => Metadata.empty
+ }
+
+ val newMetadata = new MetadataBuilder()
+ .withMetadata(metadata)
+ .putBoolean(SessionWindow.marker, true)
+ .build()
+
+ val sessionAttr = AttributeReference(
+ SESSION_COL_NAME, session.dataType, metadata = newMetadata)()
+
+ val sessionStart = PreciseTimestampConversion(session.timeColumn, TimestampType, LongType)
+ val sessionEnd = sessionStart + session.gapDuration
+
+ val literalSessionStruct = CreateNamedStruct(
+ Literal(SESSION_START) ::
+ PreciseTimestampConversion(sessionStart, LongType, TimestampType) ::
+ Literal(SESSION_END) ::
+ PreciseTimestampConversion(sessionEnd, LongType, TimestampType) ::
+ Nil)
+
+ val sessionStruct = Alias(literalSessionStruct, SESSION_COL_NAME)(
+ exprId = sessionAttr.exprId, explicitMetadata = Some(newMetadata))
+
+ val replacedPlan = p transformExpressions {
+ case s: SessionWindow => sessionAttr
+ }
+
+ // For backwards compatibility we add a filter to filter out nulls
+ val filterExpr = IsNotNull(session.timeColumn)
+
+ replacedPlan.withNewChildren(
+ Filter(filterExpr,
+ Project(sessionStruct +: child.output, child)) :: Nil)
Review comment:
https://github.com/apache/spark/pull/33367
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880796962
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/45594/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879831981
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45530/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879942790
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141013/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879900065
**[Test build #141015 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141015/testReport)** for PR 33081 at commit [`7b74ee8`](https://github.com/apache/spark/commit/7b74ee8ccbb0b8edcdb5c4631138dca3dfe9778f).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880763121
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45594/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880434824
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141042/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670174985
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
##########
@@ -335,12 +339,29 @@ abstract class SparkStrategies extends QueryPlanner[SparkPlan] {
}
}
- AggUtils.planStreamingAggregation(
- normalizedGroupingExpressions,
- aggregateExpressions.map(expr => expr.asInstanceOf[AggregateExpression]),
- rewrittenResultExpressions,
- stateVersion,
- planLater(child))
+ sessionWindowOption match {
+ case Some(sessionWindow) =>
+ val stateVersion = conf.getConf(SQLConf.STREAMING_SESSION_WINDOW_STATE_FORMAT_VERSION)
+
+ AggUtils.planStreamingAggregationForSession(
+ normalizedGroupingExpressions,
+ sessionWindow,
+ aggregateExpressions.map(expr => expr.asInstanceOf[AggregateExpression]),
+ rewrittenResultExpressions,
+ stateVersion,
+ conf.streamingSessionWindowMergeSessionInLocalPartition,
+ planLater(child))
+
+ case None =>
+ val stateVersion = conf.getConf(SQLConf.STREAMING_AGGREGATION_STATE_FORMAT_VERSION)
Review comment:
I'll remove above one. Nice finding!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-868372617
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44846/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880496509
**[Test build #141050 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141050/testReport)** for PR 33081 at commit [`1e78062`](https://github.com/apache/spark/commit/1e78062ea4f2051236ceff7d991288473ae821ce).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881215276
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45642/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r658583857
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/UpdatingSessionsIterator.scala
##########
@@ -181,7 +181,8 @@ class UpdatingSessionsIterator(
private val valueProj = GenerateUnsafeProjection.generate(valuesExpressions, inputSchema)
private val restoreProj = GenerateUnsafeProjection.generate(inputSchema,
- groupingExpressions.map(_.toAttribute) ++ valuesExpressions.map(_.toAttribute))
+ groupingWithoutSession.map(_.toAttribute) ++ Seq(sessionExpression.toAttribute) ++
Review comment:
There's a case session column is not placed at the end of the key. This line addresses such case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881538940
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141145/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #33081:
URL: https://github.com/apache/spark/pull/33081#discussion_r670214319
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/python/AggregateInPandasExec.scala
##########
@@ -53,12 +54,23 @@ case class AggregateInPandasExec(
override def producedAttributes: AttributeSet = AttributeSet(output)
- override def requiredChildDistribution: Seq[Distribution] = {
- if (groupingExpressions.isEmpty) {
- AllTuples :: Nil
- } else {
- ClusteredDistribution(groupingExpressions) :: Nil
- }
+ val sessionWindowOption = groupingExpressions.find { p =>
+ p.metadata.contains(SessionWindow.marker)
+ }
+
+ val groupingWithoutSessionExpressions = sessionWindowOption match {
+ case Some(sessionExpression) =>
+ groupingExpressions.filterNot { p => p.semanticEquals(sessionExpression) }
+
+ case None => groupingExpressions
+ }
+
+ override def requiredChildOrdering: Seq[Seq[SortOrder]] = sessionWindowOption match {
+ case Some(sessionExpression) =>
+ Seq((groupingWithoutSessionExpressions ++ Seq(sessionExpression))
+ .map(SortOrder(_, Ascending)))
+
+ case None => Seq(groupingExpressions.map(SortOrder(_, Ascending)))
Review comment:
My bad. I guess it was removed somehow during refactor. Nice finding.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879788487
**[Test build #141015 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141015/testReport)** for PR 33081 at commit [`7b74ee8`](https://github.com/apache/spark/commit/7b74ee8ccbb0b8edcdb5c4631138dca3dfe9778f).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881114900
**[Test build #141110 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141110/testReport)** for PR 33081 at commit [`4a6cff3`](https://github.com/apache/spark/commit/4a6cff3e55d91cb19bd8f512950c6ef7fb02504c).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879642812
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45513/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881218956
**[Test build #141110 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141110/testReport)** for PR 33081 at commit [`4a6cff3`](https://github.com/apache/spark/commit/4a6cff3e55d91cb19bd8f512950c6ef7fb02504c).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-879900595
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/141015/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-881383525
GA passed. Thanks! Merging to master/3.2!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #33081: [SPARK-34893][SS] Support session window natively
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #33081:
URL: https://github.com/apache/spark/pull/33081#issuecomment-880320901
**[Test build #141042 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/141042/testReport)** for PR 33081 at commit [`eef55e8`](https://github.com/apache/spark/commit/eef55e8284705f70f4dc9d08ed56fd941c98034f).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org