Skip to content

Introducing StateSchemaV3 for the TransformWithState operator with HDFSMetadataLog #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ericm-db wants to merge 9 commits into from
Closed

Introducing StateSchemaV3 for the TransformWithState operator with HDFSMetadataLog #11

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
3554af2
trying to plumb schema through planning rule
ericm-db Jun 13, 2024
2adbb02
rule is being applied
ericm-db Jun 13, 2024
f400202
adding purging
ericm-db Jun 13, 2024
45e82c9
checking if file exists in add methods
ericm-db Jun 13, 2024
95d5b2a
tests pass
ericm-db Jun 14, 2024
c383ba7
adding block to purge any metadata files that are greater than or equ…
ericm-db Jun 14, 2024
68064f1
removing duplicate code
ericm-db Jun 14, 2024
e935292
validating state variable creation
ericm-db Jun 14, 2024
52a579e
writing schema and metadata in planning rule
ericm-db Jun 14, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
adding purging
  • Loading branch information
@ericm-db
ericm-db committed Jun 13, 2024
commit f4002029c2df113bf5a5a66da0434a2f237b0e1f
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -909,10 +909,10 @@ class MicroBatchExecution(
if (shouldWriteMetadatas) {
execCtx.executionPlan.executedPlan.collect {
case tws: TransformWithStateExec =>
val schema = tws.getColumnFamilyJValue()
val metadata = tws.operatorStateMetadata()
val id = metadata.operatorInfo.operatorId
val schemaFile = stateSchemaLogs(id)
val schema = tws.getColumnFamilyJValue()
if (!schemaFile.add(execCtx.batchId, schema)) {
throw QueryExecutionErrors.concurrentStreamLogUpdate(execCtx.batchId)
}
Expand Down
2 changes: 2 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -715,6 +715,8 @@ abstract class StreamExecution(
protected def purgeOldest(): Unit = {
operatorStateMetadataLogs.foreach(
_._2.purgeOldest(minLogEntriesToMaintain))
stateSchemaLogs.foreach(
_._2.purgeOldest(minLogEntriesToMaintain))
}
}

Expand Down
1 change: 0 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -910,7 +910,6 @@ class TransformWithStateSuite extends StateStoreMetricsTest
StopStream
)

logError(s"### Restarting stream")
val result2 = inputData.toDS()
.groupByKey(x => x)
.transformWithState(
Expand Down