Spark 4.1: Add streaming-merge-append-enabled write config - #17347
Conversation
Add a Spark write configuration that makes streaming appends use the regular merge append (table.newAppend()) instead of fast append when enabled. Defaults to false, preserving existing fast-append behavior.
anuragmantri
left a comment
anuragmantri
left a comment
There was a problem hiding this comment.
The change looks good to me. Thanks @amogh-jahagirdar
| @@ -92,4 +92,8 @@ private SparkWriteOptions() {} | |||
|
|
|||
| // Controls the buffer size for variant schema inference during writes | |||
| public static final String VARIANT_INFERENCE_BUFFER_SIZE = "variant-inference-buffer-size"; | |||
|
|
|||
| // Uses the merge append instead of fast append for streaming appends | |||
| public static final String STREAMING_MERGE_APPEND_ENABLED = "streaming-merge-append-enabled"; | |||
There was a problem hiding this comment.
Curious: Should we also add a table property?
There was a problem hiding this comment.
I don't have strong opinions on this, but it seems now we move all spark specific control to https://github.com/apache/iceberg/blob/main/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTableProperties.java instead of https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/TableProperties.java in iceberg core.
There was a problem hiding this comment.
I think I'd avoid adding a table property for now. This is largely a job level configuration and under the condition of a merging writer already have the table properties for
commit.manifest.min-count-to-merge
commit.manifest-merge.enabled
Meaning in a future version if we're able to change the default to do a merging append I think we'd have all the configurations required in order to express things at a table property level.
dramaticlly
left a comment
dramaticlly
left a comment
There was a problem hiding this comment.
LGTM and suggest to reconsider the default
| @@ -92,4 +92,8 @@ private SparkWriteOptions() {} | |||
|
|
|||
| // Controls the buffer size for variant schema inference during writes | |||
| public static final String VARIANT_INFERENCE_BUFFER_SIZE = "variant-inference-buffer-size"; | |||
|
|
|||
| // Uses the merge append instead of fast append for streaming appends | |||
| public static final String STREAMING_MERGE_APPEND_ENABLED = "streaming-merge-append-enabled"; | |||
There was a problem hiding this comment.
I don't have strong opinions on this, but it seems now we move all spark specific control to https://github.com/apache/iceberg/blob/main/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTableProperties.java instead of https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/TableProperties.java in iceberg core.
|
|
||
| // Uses the merge append instead of fast append for streaming appends | ||
| public static final String STREAMING_MERGE_APPEND_ENABLED = "streaming-merge-append-enabled"; | ||
| public static final boolean STREAMING_MERGE_APPEND_ENABLED_DEFAULT = false; |
There was a problem hiding this comment.
I understand the desire to keep the original behavior, but considering the table.newAppend default to use mergeAppend, also the fact that other streaming engine integrations like Flink and Kafka Connect today use merge appends, I would be surprised to see the behavior difference in spark streaming
There was a problem hiding this comment.
I would be surprised too, and honestly I'd like to change the default here as well but I think it's best for at least 1 release to keep the behavior and then change it in the next. My rationale is mostly that it gives 1 release cycle worth of time in case there are users who somehow rely or expect there to be 1 manifest per commit (would be strange but sometimes when there's a deterministic behavior in output folks may rely on that whenever they perform some operations) or some other side effect of the fast append to be made aware of the change and have time to adjust for it.
|
What about these properties? Enabling this would mean that we just use |
Not exactly, so these properties are currently respected under the condition that a merging manifest write is already being done. So this would be operations like overwrite, rowdelta, the merging append. What enabling this conf would do is is actually cause spark streaming appends to use the merging write, and then these properties would kick in. If someone were to then configure min-count-to-merge = 50, then it would be respected. Then someone could set the table property to disable manifest merge (though not recommended here) then the merging append wouldn't merge. |
|
In this or a subsequent PR, we should update the docs. No other comments from me.Thanks @amogh-jahagirdar. |
5c3a96e to
25e6479
Compare
Done! Good call |
|
|
||
| // the second streaming append uses the regular append, which merges the two data manifests | ||
| // into a single manifest; a fast append would have left two separate manifests | ||
| table.refresh(); |
There was a problem hiding this comment.
Should the snapshot count assertion on line 201 be run after table.refresh()? (similar to the assertion on line 206)
There was a problem hiding this comment.
We actually don't even need the refresh. The test class is pinned to HadoopTable's where it'll refresh the metadata after the commit. It does rely on the TableOperations implementation but since the test class is pinned to HadoopTables, and this is consistent with other tests in this class I'll just remove the refresh.
43614e3 to
1d3083e
Compare
1d3083e to
cfa4002
Compare
|
I'll go ahead and merge. Thanks @anuragmantri @dramaticlly @szehon-ho @anoopj for reviewing! |
6 participants
Add a Spark write configuration that makes streaming appends use the regular merge append instead of fast append when enabled. Defaults to false, preserving existing fast-append behavior.
The rationale here is that as of today fast appends produces many tiny manifests, and there are Spark Streaming commit intervals where the latency of merging manifests is a negligible cost compared to the maintenance and read burden (and metadata bloat/footprint). This is a metadata equivalent to a lot of the read/write tradeoff analysis we did for DVs and equality deletes, where the notion of a "cheap write" ends up being compromised when there isn't very aggressive maintenance (which oftentimes there isn't).
Furthermore streaming engine integrations like Flink and Kafka Connect today use merge appends so I think there's general precedent that this is probably good enough for the streaming workloads Spark Streaming and Iceberg is capable of handling today, especially if it's opt in.
Especially since going forward in V4, we're addressing these cases in a much better way at the table format level (achieving fast appends without all the metadata bloat of small manifests)
I do think at some point we should flip this default (a future release version), but in the interim to prevent any unforseen regressions on upgrade, I think it makes sense to default to false.