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 2022/09/13 03:05:16 UTC
[GitHub] [spark] HyukjinKwon opened a new pull request, #37859: [SPARK-40411][SS] Refactor FlatMapGroupsWithStateExec to have a parent trait
HyukjinKwon opened a new pull request, #37859:
URL: https://github.com/apache/spark/pull/37859
### What changes were proposed in this pull request?
This PR proposes to factor the common attributes out from `FlatMapGroupsWithStateExec` to `FlatMapGroupsWithStateExecBase`.
### Why are the changes needed?
There are a lot of stuff to share if you implement another version of `FlatMapGroupsWithStateExec`.
Should better factor them out. This is also part of https://github.com/apache/spark/pull/37285 which demonstrates how the refactored trait is used.
### Does this PR introduce _any_ user-facing change?
No, this is refactoring-only.
### How was this patch tested?
Existing test cases should cover it.
--
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] HyukjinKwon closed pull request #37859: [SPARK-40411][SS] Refactor FlatMapGroupsWithStateExec to have a parent trait
Posted by GitBox <gi...@apache.org>.
HyukjinKwon closed pull request #37859: [SPARK-40411][SS] Refactor FlatMapGroupsWithStateExec to have a parent trait
URL: https://github.com/apache/spark/pull/37859
--
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] HyukjinKwon commented on pull request #37859: [SPARK-40411][SS] Refactor FlatMapGroupsWithStateExec to have a parent trait
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #37859:
URL: https://github.com/apache/spark/pull/37859#issuecomment-1244947785
Let me get this in - I am pretty much confident that this is the right change too :-).
Merged to master.
--
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] HyukjinKwon commented on a diff in pull request #37859: [SPARK-40411][SS] Refactor FlatMapGroupsWithStateExec to have a parent trait
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #37859:
URL: https://github.com/apache/spark/pull/37859#discussion_r969127821
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FlatMapGroupsWithStateExec.scala:
##########
@@ -33,59 +34,35 @@ import org.apache.spark.util.{CompletionIterator, SerializableConfiguration}
/**
* Physical operator for executing `FlatMapGroupsWithState`
- *
- * @param func function called on each group
- * @param keyDeserializer used to extract the key object for each group.
- * @param valueDeserializer used to extract the items in the iterator from an input row.
- * @param initialStateDeserializer used to extract the state object from the initialState dataset
- * @param groupingAttributes used to group the data
- * @param dataAttributes used to read the data
- * @param outputObjAttr Defines the output object
- * @param stateEncoder used to serialize/deserialize state before calling `func`
- * @param outputMode the output mode of `func`
- * @param timeoutConf used to timeout groups that have not received data in a while
- * @param batchTimestampMs processing timestamp of the current batch.
- * @param eventTimeWatermark event time watermark for the current batch
- * @param initialState the user specified initial state
- * @param hasInitialState indicates whether the initial state is provided or not
- * @param child the physical plan for the underlying data
*/
-case class FlatMapGroupsWithStateExec(
- func: (Any, Iterator[Any], LogicalGroupState[Any]) => Iterator[Any],
- keyDeserializer: Expression,
- valueDeserializer: Expression,
- initialStateDeserializer: Expression,
- groupingAttributes: Seq[Attribute],
- initialStateGroupAttrs: Seq[Attribute],
- dataAttributes: Seq[Attribute],
- initialStateDataAttrs: Seq[Attribute],
- outputObjAttr: Attribute,
- stateInfo: Option[StatefulOperatorStateInfo],
- stateEncoder: ExpressionEncoder[Any],
- stateFormatVersion: Int,
- outputMode: OutputMode,
- timeoutConf: GroupStateTimeout,
- batchTimestampMs: Option[Long],
- eventTimeWatermark: Option[Long],
- initialState: SparkPlan,
- hasInitialState: Boolean,
- child: SparkPlan
- ) extends BinaryExecNode with ObjectProducerExec with StateStoreWriter with WatermarkSupport {
-
- import FlatMapGroupsWithStateExecHelper._
+trait FlatMapGroupsWithStateExecBase
+ extends StateStoreWriter with WatermarkSupport {
import GroupStateImpl._
+ import FlatMapGroupsWithStateExecHelper._
- override def left: SparkPlan = child
+ protected val groupingAttributes: Seq[Attribute]
- override def right: SparkPlan = initialState
+ protected val initialStateDeserializer: Expression
+ protected val initialStateGroupAttrs: Seq[Attribute]
+ protected val initialStateDataAttrs: Seq[Attribute]
+ protected val initialState: SparkPlan
+ protected val hasInitialState: Boolean
- private val isTimeoutEnabled = timeoutConf != NoTimeout
- private val watermarkPresent = child.output.exists {
+ val stateInfo: Option[StatefulOperatorStateInfo]
Review Comment:
I remember I exposed them for a reason at that time.. let me change this to `protected` and see if it passes.
--
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 diff in pull request #37859: [SPARK-40411][SS] Refactor FlatMapGroupsWithStateExec to have a parent trait
Posted by GitBox <gi...@apache.org>.
viirya commented on code in PR #37859:
URL: https://github.com/apache/spark/pull/37859#discussion_r969213274
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FlatMapGroupsWithStateExec.scala:
##########
@@ -33,59 +34,35 @@ import org.apache.spark.util.{CompletionIterator, SerializableConfiguration}
/**
* Physical operator for executing `FlatMapGroupsWithState`
Review Comment:
It's nice if we could update this doc too.
--
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 diff in pull request #37859: [SPARK-40411][SS] Refactor FlatMapGroupsWithStateExec to have a parent trait
Posted by GitBox <gi...@apache.org>.
viirya commented on code in PR #37859:
URL: https://github.com/apache/spark/pull/37859#discussion_r969216072
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FlatMapGroupsWithStateExec.scala:
##########
@@ -252,31 +229,25 @@ case class FlatMapGroupsWithStateExec(
session.sqlContext.sessionState,
Some(session.sqlContext.streams.stateStoreCoordinator)
) { case (store: StateStore, singleIterator: Iterator[InternalRow]) =>
- val processor = new InputProcessor(store)
+ val processor = createInputProcessor(store)
processDataWithPartition(singleIterator, store, processor)
}
}
}
- /** Helper class to update the state store */
- class InputProcessor(store: StateStore) {
+ def createInputProcessor(store: StateStore): InputProcessor
- // Converters for translating input keys, values, output data between rows and Java objects
- private val getKeyObj =
- ObjectOperator.deserializeRowToObject(keyDeserializer, groupingAttributes)
- private val getValueObj =
- ObjectOperator.deserializeRowToObject(valueDeserializer, dataAttributes)
- private val getOutputRow = ObjectOperator.wrapObjectToRow(outputObjectType)
+ abstract class InputProcessor(store: StateStore) {
Review Comment:
Maybe good to have some doc for this abstract class?
--
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] HyukjinKwon commented on a diff in pull request #37859: [SPARK-40411][SS] Refactor FlatMapGroupsWithStateExec to have a parent trait
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #37859:
URL: https://github.com/apache/spark/pull/37859#discussion_r969129411
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FlatMapGroupsWithStateExec.scala:
##########
@@ -33,59 +34,35 @@ import org.apache.spark.util.{CompletionIterator, SerializableConfiguration}
/**
* Physical operator for executing `FlatMapGroupsWithState`
- *
- * @param func function called on each group
- * @param keyDeserializer used to extract the key object for each group.
- * @param valueDeserializer used to extract the items in the iterator from an input row.
- * @param initialStateDeserializer used to extract the state object from the initialState dataset
- * @param groupingAttributes used to group the data
- * @param dataAttributes used to read the data
- * @param outputObjAttr Defines the output object
- * @param stateEncoder used to serialize/deserialize state before calling `func`
- * @param outputMode the output mode of `func`
- * @param timeoutConf used to timeout groups that have not received data in a while
- * @param batchTimestampMs processing timestamp of the current batch.
- * @param eventTimeWatermark event time watermark for the current batch
- * @param initialState the user specified initial state
- * @param hasInitialState indicates whether the initial state is provided or not
- * @param child the physical plan for the underlying data
*/
-case class FlatMapGroupsWithStateExec(
- func: (Any, Iterator[Any], LogicalGroupState[Any]) => Iterator[Any],
- keyDeserializer: Expression,
- valueDeserializer: Expression,
- initialStateDeserializer: Expression,
- groupingAttributes: Seq[Attribute],
- initialStateGroupAttrs: Seq[Attribute],
- dataAttributes: Seq[Attribute],
- initialStateDataAttrs: Seq[Attribute],
- outputObjAttr: Attribute,
- stateInfo: Option[StatefulOperatorStateInfo],
- stateEncoder: ExpressionEncoder[Any],
- stateFormatVersion: Int,
- outputMode: OutputMode,
- timeoutConf: GroupStateTimeout,
- batchTimestampMs: Option[Long],
- eventTimeWatermark: Option[Long],
- initialState: SparkPlan,
- hasInitialState: Boolean,
- child: SparkPlan
- ) extends BinaryExecNode with ObjectProducerExec with StateStoreWriter with WatermarkSupport {
-
- import FlatMapGroupsWithStateExecHelper._
+trait FlatMapGroupsWithStateExecBase
+ extends StateStoreWriter with WatermarkSupport {
import GroupStateImpl._
+ import FlatMapGroupsWithStateExecHelper._
- override def left: SparkPlan = child
+ protected val groupingAttributes: Seq[Attribute]
- override def right: SparkPlan = initialState
+ protected val initialStateDeserializer: Expression
+ protected val initialStateGroupAttrs: Seq[Attribute]
+ protected val initialStateDataAttrs: Seq[Attribute]
+ protected val initialState: SparkPlan
+ protected val hasInitialState: Boolean
- private val isTimeoutEnabled = timeoutConf != NoTimeout
- private val watermarkPresent = child.output.exists {
+ val stateInfo: Option[StatefulOperatorStateInfo]
Review Comment:
The error is like this:
```
[error] /.../spark/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FlatMapGroupsWithStateExec.scala:51:17: overriding method stateInfo in trait StatefulOperator of type => Option[org.apache.spark.sql.execution.streaming.StatefulOperatorStateInfo];
[error] value stateInfo has weaker access privileges; it should be public
[error] protected val stateInfo: Option[StatefulOperatorStateInfo]
[error] ^
[error] /.../spark/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FlatMapGroupsWithStateExec.scala:57:17: overriding method eventTimeWatermark in trait WatermarkSupport of type => Option[Long];
[error] value eventTimeWatermark has weaker access privileges; it should be public
[error] protected val eventTimeWatermark: Option[Long]
[error] ^
[error] two errors found
[error] (sql / Compile / compileIncremental) Compilation failed
[error] Total time: 160 s (02:40), completed Sep 13, 2022 1:03:01 PM
```
--
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] HyukjinKwon commented on pull request #37859: [SPARK-40411][SS] Refactor FlatMapGroupsWithStateExec to have a parent trait
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #37859:
URL: https://github.com/apache/spark/pull/37859#issuecomment-1244837063
cc @HeartSaVioR and @viirya FYI
--
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 diff in pull request #37859: [SPARK-40411][SS] Refactor FlatMapGroupsWithStateExec to have a parent trait
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on code in PR #37859:
URL: https://github.com/apache/spark/pull/37859#discussion_r969132744
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FlatMapGroupsWithStateExec.scala:
##########
@@ -33,59 +34,35 @@ import org.apache.spark.util.{CompletionIterator, SerializableConfiguration}
/**
* Physical operator for executing `FlatMapGroupsWithState`
- *
- * @param func function called on each group
- * @param keyDeserializer used to extract the key object for each group.
- * @param valueDeserializer used to extract the items in the iterator from an input row.
- * @param initialStateDeserializer used to extract the state object from the initialState dataset
- * @param groupingAttributes used to group the data
- * @param dataAttributes used to read the data
- * @param outputObjAttr Defines the output object
- * @param stateEncoder used to serialize/deserialize state before calling `func`
- * @param outputMode the output mode of `func`
- * @param timeoutConf used to timeout groups that have not received data in a while
- * @param batchTimestampMs processing timestamp of the current batch.
- * @param eventTimeWatermark event time watermark for the current batch
- * @param initialState the user specified initial state
- * @param hasInitialState indicates whether the initial state is provided or not
- * @param child the physical plan for the underlying data
*/
-case class FlatMapGroupsWithStateExec(
- func: (Any, Iterator[Any], LogicalGroupState[Any]) => Iterator[Any],
- keyDeserializer: Expression,
- valueDeserializer: Expression,
- initialStateDeserializer: Expression,
- groupingAttributes: Seq[Attribute],
- initialStateGroupAttrs: Seq[Attribute],
- dataAttributes: Seq[Attribute],
- initialStateDataAttrs: Seq[Attribute],
- outputObjAttr: Attribute,
- stateInfo: Option[StatefulOperatorStateInfo],
- stateEncoder: ExpressionEncoder[Any],
- stateFormatVersion: Int,
- outputMode: OutputMode,
- timeoutConf: GroupStateTimeout,
- batchTimestampMs: Option[Long],
- eventTimeWatermark: Option[Long],
- initialState: SparkPlan,
- hasInitialState: Boolean,
- child: SparkPlan
- ) extends BinaryExecNode with ObjectProducerExec with StateStoreWriter with WatermarkSupport {
-
- import FlatMapGroupsWithStateExecHelper._
+trait FlatMapGroupsWithStateExecBase
+ extends StateStoreWriter with WatermarkSupport {
import GroupStateImpl._
+ import FlatMapGroupsWithStateExecHelper._
- override def left: SparkPlan = child
+ protected val groupingAttributes: Seq[Attribute]
- override def right: SparkPlan = initialState
+ protected val initialStateDeserializer: Expression
+ protected val initialStateGroupAttrs: Seq[Attribute]
+ protected val initialStateDataAttrs: Seq[Attribute]
+ protected val initialState: SparkPlan
+ protected val hasInitialState: Boolean
- private val isTimeoutEnabled = timeoutConf != NoTimeout
- private val watermarkPresent = child.output.exists {
+ val stateInfo: Option[StatefulOperatorStateInfo]
Review Comment:
Ah OK that was from another trait. Thanks for explanation.
--
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