HyukjinKwon commented on code in PR #37893:
URL: https://github.com/apache/spark/pull/37893#discussion_r973844153
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2705,6 +2705,44 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val MAP_PANDAS_UDF_WITH_STATE_SOFT_LIMIT_SIZE_PER_BATCH =
+
buildConf("spark.sql.execution.applyInPandasWithState.softLimitSizePerBatch")
+ .internal()
+ .doc("When using applyInPandasWithState, set a soft limit of the
accumulated size of " +
+ "records that can be written to a single ArrowRecordBatch in memory.
This is used to " +
+ "restrict the amount of memory being used to materialize the data in
both executor and " +
+ "Python worker. The accumulated size of records are calculated via
sampling a set of " +
+ "records. Splitting the ArrowRecordBatch is performed per record, so
unless a record " +
+ "is quite huge, the size of constructed ArrowRecordBatch will be
around the " +
+ "configured value.")
+ .version("3.4.0")
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefaultString("64MB")
+
+ val MAP_PANDAS_UDF_WITH_STATE_MIN_DATA_COUNT_FOR_SAMPLE =
+
buildConf("spark.sql.execution.applyInPandasWithState.minDataCountForSample")
+ .internal()
+ .doc("When using applyInPandasWithState, specify the minimum number of
records to sample " +
+ "the size of record. The size being retrieved from sampling will be
used to estimate " +
+ "the accumulated size of records. Note that limiting by size does not
work if the " +
+ "number of records are less than the configured value. For such case,
ArrowRecordBatch " +
+ "will only be split for soft timeout.")
+ .version("3.4.0")
+ .intConf
+ .createWithDefault(100)
+
+ val MAP_PANDAS_UDF_WITH_STATE_SOFT_TIMEOUT_PURGE_BATCH =
+
buildConf("spark.sql.execution.applyInPandasWithState.softTimeoutPurgeBatch")
+ .internal()
+ .doc("When using applyInPandasWithState, specify the soft timeout for
purging the " +
+ "ArrowRecordBatch. If batching records exceeds the timeout, Spark will
force splitting " +
+ "the ArrowRecordBatch regardless of estimated size. This config
ensures the receiver " +
+ "of data (both executor and Python worker) to not wait indefinitely
for sender to " +
+ "complete the ArrowRecordBatch, which may hurt both throughput and
latency.")
+ .version("3.4.0")
+ .timeConf(TimeUnit.MILLISECONDS)
+ .createWithDefaultString("100ms")
Review Comment:
For this, can we just leverage `spark.sql.execution.pandas.udf.buffer.size`
(the feature this PR adds already respects it) if the flush time matters? That
configuration is for the purpose.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnsupportedOperationChecker.scala:
##########
@@ -142,6 +143,17 @@ object UnsupportedOperationChecker extends Logging {
" or the output mode is not append on a streaming
DataFrames/Datasets")(plan)
}
+ val applyInPandasWithStates = plan.collect {
+ case f: FlatMapGroupsInPandasWithState if f.isStreaming => f
+ }
+
+ // Disallow multiple `applyInPandasWithState`s.
+ if (applyInPandasWithStates.size >= 2) {
Review Comment:
no biggie but ..
```suggestion
if (applyInPandasWithStates.size > 1) {
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2705,6 +2705,44 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val MAP_PANDAS_UDF_WITH_STATE_SOFT_LIMIT_SIZE_PER_BATCH =
+
buildConf("spark.sql.execution.applyInPandasWithState.softLimitSizePerBatch")
+ .internal()
+ .doc("When using applyInPandasWithState, set a soft limit of the
accumulated size of " +
+ "records that can be written to a single ArrowRecordBatch in memory.
This is used to " +
+ "restrict the amount of memory being used to materialize the data in
both executor and " +
+ "Python worker. The accumulated size of records are calculated via
sampling a set of " +
+ "records. Splitting the ArrowRecordBatch is performed per record, so
unless a record " +
+ "is quite huge, the size of constructed ArrowRecordBatch will be
around the " +
+ "configured value.")
+ .version("3.4.0")
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefaultString("64MB")
Review Comment:
I think we should have a general configuration for this later that applies
to all Arrow batch (SPARK-23258). I think we should reuse
`spark.sql.execution.arrow.maxRecordsPerBatch` for the time being.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/ApplyInPandasWithStatePythonRunner.scala:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.execution.python
+
+import java.io._
+
+import scala.collection.JavaConverters._
+
+import org.apache.arrow.vector.VectorSchemaRoot
+import org.apache.arrow.vector.ipc.ArrowStreamWriter
+import org.json4s._
+import org.json4s.jackson.JsonMethods._
+
+import org.apache.spark.api.python._
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.api.python.PythonSQLUtils
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow
+import
org.apache.spark.sql.execution.python.ApplyInPandasWithStatePythonRunner.{InType,
OutType, OutTypeForState, STATE_METADATA_SCHEMA_FROM_PYTHON_WORKER}
+import
org.apache.spark.sql.execution.python.ApplyInPandasWithStateWriter.STATE_METADATA_SCHEMA
+import org.apache.spark.sql.execution.streaming.GroupStateImpl
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.vectorized.{ArrowColumnVector, ColumnarBatch}
+
+
+/**
+ * [[ArrowPythonRunner]] with [[org.apache.spark.sql.streaming.GroupState]].
+ */
+class ApplyInPandasWithStatePythonRunner(
+ funcs: Seq[ChainedPythonFunctions],
+ evalType: Int,
+ argOffsets: Array[Array[Int]],
+ inputSchema: StructType,
+ override protected val timeZoneId: String,
+ initialWorkerConf: Map[String, String],
+ stateEncoder: ExpressionEncoder[Row],
+ keySchema: StructType,
+ valueSchema: StructType,
+ stateValueSchema: StructType,
+ softLimitBytesPerBatch: Long,
+ minDataCountForSample: Int,
+ softTimeoutMillsPurgeBatch: Long)
+ extends BasePythonRunner[InType, OutType](funcs, evalType, argOffsets)
+ with PythonArrowInput[InType]
+ with PythonArrowOutput[OutType] {
+
+ override protected val schema: StructType = inputSchema.add("!__state__!",
STATE_METADATA_SCHEMA)
Review Comment:
I suspect it's using `!` here because `!` cannot be an identifier in Spark
SQL (?). To be absolutely strict, such column names are allowed in some places
of DataFrame API (e.g, `spark.range(1).toDF("!__state__!")`).
I believe we use internal column names such `__grouping__id`,
`__file_source_metadata_col`, `__metadata_col` and `_groupingexpression` in
general. We're retrieving them positionally in Python worker side so I assume
this is fine to have a duplicate name ...
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/FlatMapGroupsInPandasWithStateExec.scala:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.execution.python
+
+import org.apache.spark.TaskContext
+import org.apache.spark.api.python.{ChainedPythonFunctions, PythonEvalType}
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.encoders.{ExpressionEncoder, RowEncoder}
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.{EventTimeTimeout,
ProcessingTimeTimeout}
+import org.apache.spark.sql.catalyst.plans.physical.Distribution
+import org.apache.spark.sql.execution.{GroupedIterator, SparkPlan,
UnaryExecNode}
+import org.apache.spark.sql.execution.python.PandasGroupUtils.resolveArgOffsets
+import org.apache.spark.sql.execution.streaming._
+import org.apache.spark.sql.execution.streaming.GroupStateImpl.NO_TIMESTAMP
+import
org.apache.spark.sql.execution.streaming.state.FlatMapGroupsWithStateExecHelper.StateData
+import org.apache.spark.sql.execution.streaming.state.StateStore
+import org.apache.spark.sql.streaming.{GroupStateTimeout, OutputMode}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.ArrowUtils
+import org.apache.spark.util.CompletionIterator
+
+/**
+ * Physical operator for executing
+ *
[[org.apache.spark.sql.catalyst.plans.logical.FlatMapGroupsInPandasWithState]]
+ *
+ * @param functionExpr function called on each group
+ * @param groupingAttributes used to group the data
+ * @param outAttributes used to define the output rows
+ * @param stateType used to serialize/deserialize state before calling
`functionExpr`
+ * @param stateInfo `StatefulOperatorStateInfo` to identify the state store
for a given operator.
+ * @param stateFormatVersion the version of state format.
+ * @param outputMode the output mode of `functionExpr`
+ * @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 child logical plan of the underlying data
+ */
+case class FlatMapGroupsInPandasWithStateExec(
+ functionExpr: Expression,
+ groupingAttributes: Seq[Attribute],
+ outAttributes: Seq[Attribute],
+ stateType: StructType,
+ stateInfo: Option[StatefulOperatorStateInfo],
+ stateFormatVersion: Int,
+ outputMode: OutputMode,
+ timeoutConf: GroupStateTimeout,
+ batchTimestampMs: Option[Long],
+ eventTimeWatermark: Option[Long],
+ child: SparkPlan) extends UnaryExecNode with
FlatMapGroupsWithStateExecBase {
+
+ // TODO(SPARK-40444): Add the support of initial state.
+ override protected val initialStateDeserializer: Expression = null
+ override protected val initialStateGroupAttrs: Seq[Attribute] = null
+ override protected val initialStateDataAttrs: Seq[Attribute] = null
+ override protected val initialState: SparkPlan = null
Review Comment:
```suggestion
override protected val initialStateDeserializer: Expression = _
override protected val initialStateGroupAttrs: Seq[Attribute] = _
override protected val initialStateDataAttrs: Seq[Attribute] = _
override protected val initialState: SparkPlan = _
```
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
