ca2ba4fe64
### What changes were proposed in this pull request?
This patch fixes the behavior of ProgressReporter which always overwrite the value of "updated" of state operator to 0 if there's no new data. The behavior is correct only when we copy the state progress from "previous" executed plan, meaning no batch has been run. (Nonzero value of "updated" would be odd if batch didn't run, so it was correct.)
It was safe to assume no data is no batch, but SPARK-24156 enables empty data can run the batch if Spark needs to deal with watermark. After the patch, it only overwrites the value if both two conditions are met: 1) no data 2) no batch.
### Why are the changes needed?
Currently Spark doesn't reflect correct metrics when empty batch is run and this patch fixes it.
### Does this PR introduce any user-facing change?
No.
### How was this patch tested?
Modified UT. Note that FlatMapGroupsWithState increases the value of "updated" when state rows are removed.
Also manually tested via below query (not a simple query to test with spark-shell, as you'll meet closure issue in spark-shell while playing with state func):
> query
```
case class RunningCount(count: Long)
object TestFlatMapGroupsWithState {
def main(args: Array[String]): Unit = {
import org.apache.spark.sql.SparkSession
val ss = SparkSession
.builder()
.appName("TestFlatMapGroupsWithState")
.getOrCreate()
ss.conf.set("spark.sql.shuffle.partitions", "5")
import ss.implicits._
val stateFunc = (key: String, values: Iterator[String], state: GroupState[RunningCount]) => {
if (state.hasTimedOut) {
// End users are not restricted to remove the state here - they can update the
// state as well. For example, event time session window would have list of
// sessions here and it cannot remove entire state.
state.update(RunningCount(-1))
Iterator((key, "-1"))
} else {
val count = state.getOption.map(_.count).getOrElse(0L) + values.size
state.update(RunningCount(count))
state.setTimeoutDuration("1 seconds")
Iterator((key, count.toString))
}
}
implicit val sqlContext = ss.sqlContext
val inputData = MemoryStream[String]
val result = inputData
.toDF()
.as[String]
.groupByKey { v => v }
.flatMapGroupsWithState(OutputMode.Append(), GroupStateTimeout.ProcessingTimeTimeout())(stateFunc)
val query = result
.writeStream
.format("memory")
.option("queryName", "test")
.outputMode("append")
.trigger(Trigger.ProcessingTime("5 second"))
.start()
Thread.sleep(1000)
var chIdx: Long = 0
while (true) {
(chIdx to chIdx + 4).map { idx => inputData.addData(idx.toString) }
chIdx += 5
// intentionally sleep much more than trigger to enable "empty" batch
Thread.sleep(10 * 1000)
}
}
}
```
> before the patch (batch 3 which was an "empty" batch)
```
{
"id":"de945a5c-882b-4dae-aa58-cb8261cbaf9e",
"runId":"f1eb6d0d-3cd5-48b2-a03b-5e989b6c151b",
"name":"test",
"timestamp":"2019-11-18T07:00:25.005Z",
"batchId":3,
"numInputRows":0,
"inputRowsPerSecond":0.0,
"processedRowsPerSecond":0.0,
"durationMs":{
"addBatch":1664,
"getBatch":0,
"latestOffset":0,
"queryPlanning":29,
"triggerExecution":1789,
"walCommit":51
},
"stateOperators":[
{
"numRowsTotal":10,
"numRowsUpdated":0,
"memoryUsedBytes":5130,
"customMetrics":{
"loadedMapCacheHitCount":15,
"loadedMapCacheMissCount":0,
"stateOnCurrentVersionSizeBytes":2722
}
}
],
"sources":[
{
"description":"MemoryStream[value#1]",
"startOffset":9,
"endOffset":9,
"numInputRows":0,
"inputRowsPerSecond":0.0,
"processedRowsPerSecond":0.0
}
],
"sink":{
"description":"MemorySink",
"numOutputRows":5
}
}
```
> after the patch (batch 3 which was an "empty" batch)
```
{
"id":"7cb41623-6b9a-408e-ae02-6796ec636fa0",
"runId":"17847710-ddfe-45f5-a7ab-b160e149382f",
"name":"test",
"timestamp":"2019-11-18T07:02:25.005Z",
"batchId":3,
"numInputRows":0,
"inputRowsPerSecond":0.0,
"processedRowsPerSecond":0.0,
"durationMs":{
"addBatch":1196,
"getBatch":0,
"latestOffset":0,
"queryPlanning":30,
"triggerExecution":1333,
"walCommit":46
},
"stateOperators":[
{
"numRowsTotal":10,
"numRowsUpdated":5,
"memoryUsedBytes":5130,
"customMetrics":{
"loadedMapCacheHitCount":15,
"loadedMapCacheMissCount":0,
"stateOnCurrentVersionSizeBytes":2722
}
}
],
"sources":[
{
"description":"MemoryStream[value#1]",
"startOffset":9,
"endOffset":9,
"numInputRows":0,
"inputRowsPerSecond":0.0,
"processedRowsPerSecond":0.0
}
],
"sink":{
"description":"MemorySink",
"numOutputRows":5
}
}
```
"numRowsUpdated" is `0` in "stateOperators" before applying the patch which is "wrong", as we "update" the state when timeout occurs. After applying the patch, it correctly represents the "numRowsUpdated" as `5` in "stateOperators".
Closes #25987 from HeartSaVioR/SPARK-29314.
Authored-by: Jungtaek Lim (HeartSaVioR) <kabhwan.opensource@gmail.com>
Signed-off-by: Burak Yavuz <brkyvz@gmail.com>
|
||
|---|---|---|
| .. | ||
| catalyst | ||
| core | ||
| hive | ||
| hive-thriftserver | ||
| create-docs.sh | ||
| gen-sql-api-docs.py | ||
| gen-sql-config-docs.py | ||
| mkdocs.yml | ||
| README.md | ||
Spark SQL
This module provides support for executing relational queries expressed in either SQL or the DataFrame/Dataset API.
Spark SQL is broken up into four subprojects:
- Catalyst (sql/catalyst) - An implementation-agnostic framework for manipulating trees of relational operators and expressions.
- Execution (sql/core) - A query planner / execution engine for translating Catalyst's logical query plans into Spark RDDs. This component also includes a new public interface, SQLContext, that allows users to execute SQL or LINQ statements against existing RDDs and Parquet files.
- Hive Support (sql/hive) - Includes extensions that allow users to write queries using a subset of HiveQL and access data from a Hive Metastore using Hive SerDes. There are also wrappers that allow users to run queries that include Hive UDFs, UDAFs, and UDTFs.
- HiveServer and CLI support (sql/hive-thriftserver) - Includes support for the SQL CLI (bin/spark-sql) and a HiveServer2 (for JDBC/ODBC) compatible server.
Running ./sql/create-docs.sh generates SQL documentation for built-in functions under sql/site, and SQL configuration documentation that gets included as part of configuration.md in the main docs directory.