Repository navigation
Docs: Add docs for Table Maintenance in Flink - #13853
Conversation
|
Hello, I have submitted a new PR with the previous feedback incorporated as much as possible. Since this is my first time contributing to the Docs, I am still learning and going through some trial and error. I really appreciate all your guidance and support. I will gladly incorporate any additional feedback you may have. If everything looks good, I would appreciate it if you could merge the PR. Thank you very much. |
|
|
||
| ## Overview | ||
|
|
||
| In **Apache Iceberg** deployments within **Flink streaming environments**, implementing automated table maintenance operations—including `snapshot expiration`, `small file compaction`, and `orphan file cleanup`—is critical for optimal performance and storage efficiency. |
There was a problem hiding this comment.
optimal performance -> optimal query performance
There was a problem hiding this comment.
Let's see if we can merge the orphan file cleanup in 1.10.0.
- If yes, then we need to mentioned it in the features
- If no, then we should remove it from here
|
|
||
| Example usage: | ||
| ```java | ||
| // Snapshot expiration | ||
| .add(ExpireSnapshots.builder() | ||
| .maxSnapshotAge(Duration.ofDays(7)) | ||
| .retainLast(10) | ||
| .deleteBatchSize(1000)) | ||
|
|
||
| // Continuous file optimization | ||
| .add(RewriteDataFiles.builder() | ||
| .targetFileSizeBytes(256 * 1024 * 1024) | ||
| .minFileSizeBytes(32 * 1024 * 1024) | ||
| .partialProgressEnabled(true) | ||
| .partialProgressMaxCommits(5)) | ||
| ``` |
There was a problem hiding this comment.
Do we need this example here? We have more detailed examples below.
If we decide to need this, we might want to separate the ExpireSnapshots and the RewriteDataFiles and put them after the descriptions above
|
I have incorporated the feedback you provided. I would appreciate it if you could review it. Where can I check the addition of orphan file cleanup in version 1.10.0? |
|
Adding to the docs is more challenging than I expected, but also really fun. I hope it gets merged soon :D |
| ```java | ||
| TriggerLockFactory lockFactory = new ZkLockFactory( | ||
| "localhost:2181", // ZooKeeper connection string | ||
| "catalog.db.table" // Lock ID (unique identifier) |
There was a problem hiding this comment.
We also need sessionTimeoutMs,connectionTimeoutMs,baseSleepTimeMs and maxRetries in Constructor.
|
|
||
| | Method | Description | Default Value | Type | | ||
| |--------|-------------|---------------|------| | ||
| | `maxSnapshotAge(Duration)` | Maximum age of snapshots to retain | None | Duration | |
There was a problem hiding this comment.
Default value is 5 days.
| | Method | Description | Default Value | Type | | ||
| |--------|-------------|---------------|------| | ||
| | `maxSnapshotAge(Duration)` | Maximum age of snapshots to retain | None | Duration | | ||
| | `retainLast(int)` | Minimum number of snapshots to retain | None | int | |
| | `retainLast(int)` | Minimum number of snapshots to retain | None | int | | ||
| | `deleteBatchSize(int)` | Number of files to delete in each batch | 1000 | int | | ||
| | `planningWorkerPoolSize(int)` | Number of worker threads for planning snapshot expiration | Shared worker pool | int | | ||
| | `cleanExpiredMetadata(boolean)` | Remove expired metadata files when expiring snapshots | true | boolean | |
There was a problem hiding this comment.
I think in this place default value is false.
| | `targetFileSizeBytes(long)` | Target size for rewritten files | Table property or 512MB | long | | ||
| | `minFileSizeBytes(long)` | Minimum size of files eligible for compaction | 0 | long | | ||
| | `maxFileSizeBytes(long)` | Maximum size of files eligible for compaction | Long.MAX_VALUE | long | | ||
| | `minInputFiles(int)` | Minimum number of files to trigger rewrite | Integer.MAX_VALUE | int | |
There was a problem hiding this comment.
I think the default value is 5.
| | `maxFileGroupSizeBytes(long)` | Maximum total size of a file group | 107374182400 (100GB) | long | | ||
| | `maxFilesToRewrite(int)` | Maximum number of files to rewrite per task | Integer.MAX_VALUE | int | | ||
| | `partialProgressEnabled(boolean)` | Enable partial progress commits | false | boolean | | ||
| | `partialProgressMaxCommits(int)` | Maximum commits for partial progress | 10 | int | |
There was a problem hiding this comment.
Maximum commits for partial progress, if partial progress is enabled
|
I reviewed the classes carefully to check the default values, but I got quite a few wrong. My apologies. For I’ve also made other changes based on your feedback. Thank you for your review. |
|
I’ve updated the document according to your feedback. The Default Lock section was incorrectly added, so I’ve removed it. Thank you for your review. |
| ); | ||
|
|
||
| // Set up maintenance with comprehensive configuration | ||
| TableMaintenance.forTable(env, tableLoader, TriggerLockFactory.defaultLockFactory()) |
There was a problem hiding this comment.
TriggerLockFactory.defaultLockFactory() We don't have this method. This part should be changed.
| ``` | ||
|
|
||
| #### ZooKeeper Lock Factory | ||
| Uses Apache ZooKeeper for distributed coordination: |
There was a problem hiding this comment.
| Uses Apache ZooKeeper for distributed coordination: | |
| Uses Apache ZooKeeper for distributed locks: |
| | Method | Description | Default Value | Type | | ||
| |--------|-------------|---------------|------| | ||
| | `targetFileSizeBytes(long)` | Target size for rewritten files | Table property or 512MB | long | | ||
| | `minFileSizeBytes(long)` | Minimum size of files eligible for compaction | 0 | long | |
There was a problem hiding this comment.
Default value is 75% of target file size
| |--------|-------------|---------------|------| | ||
| | `targetFileSizeBytes(long)` | Target size for rewritten files | Table property or 512MB | long | | ||
| | `minFileSizeBytes(long)` | Minimum size of files eligible for compaction | 0 | long | | ||
| | `maxFileSizeBytes(long)` | Maximum size of files eligible for compaction | Long.MAX_VALUE | long | |
There was a problem hiding this comment.
Default value is 180% of target file size
| | `targetFileSizeBytes(long)` | Target size for rewritten files | Table property or 512MB | long | | ||
| | `minFileSizeBytes(long)` | Minimum size of files eligible for compaction | 0 | long | | ||
| | `maxFileSizeBytes(long)` | Maximum size of files eligible for compaction | Long.MAX_VALUE | long | | ||
| | `minInputFiles(int)` | Minimum number of files to trigger rewrite | 1 | int | |
| | `deleteFileThreshold(int)` | Minimum delete-file count per data file to force rewrite | Integer.MAX_VALUE | int | | ||
| | `rewriteAll(boolean)` | Rewrite all data files regardless of thresholds | false | boolean | | ||
| | `maxFileGroupSizeBytes(long)` | Maximum total size of a file group | 107374182400 (100GB) | long | | ||
| | `maxFilesToRewrite(int)` | Maximum number of files to rewrite per task | Integer.MAX_VALUE | int | |
There was a problem hiding this comment.
Default value is null. About the description we can add If this option is not specified, all eligible files will be rewritten.
| 'namespace' = 'db', | ||
| 'table' = 'tbl', |
There was a problem hiding this comment.
Key should be catalog-database and catalog-table
| jdbcProps // JDBC connection properties | ||
| ); | ||
| ``` | ||
|
|
There was a problem hiding this comment.
Can we add some specific configuration item explanations, such as INIT_LOCK_TABLES_PROPERTY, since this configuration item is mentioned in the SQL part below but does not appear in the Java part?
| - Adjust `deleteBatchSize` based on storage performance | ||
| - Enable `partialProgressEnabled` for large rewrite operations | ||
| - Set reasonable `maxRewriteBytes` limits | ||
|
|
There was a problem hiding this comment.
May be we can add Setting an appropriate maxFileGroupSizeBytes can break down large FileGroups into smaller ones, thereby increasing the speed of parallel processing.
| | `partialProgressEnabled(boolean)` | Enable partial progress commits | false | boolean | | ||
| | `partialProgressMaxCommits(int)` | Maximum commits allowed for partial progress when partialProgressEnabled is true | 10 | int | | ||
| | `maxRewriteBytes(long)` | Maximum bytes to rewrite per execution | Long.MAX_VALUE | long | | ||
|
|
There was a problem hiding this comment.
Miss a method filter
|
Hello, I’ve incorporated the feedback you mentioned above. This is my first time writing a Docs from start to finish, so while I’m enjoying it, I also feel like I might be causing some trouble, haha... That said, please continue to provide feedback. I really appreciate it! :D |
Guosmilesmile
left a comment
There was a problem hiding this comment.
I have left some mirror comments. I think we can reuse the existing page https://iceberg.apache.org/docs/nightly/flink-actions/ instead of adding a new file. We can rename flink-actions to flink-maintenance, which would eliminate the need for a new page.
| | `partialProgressEnabled(boolean)` | Enable partial progress commits | false | boolean | | ||
| | `partialProgressMaxCommits(int)` | Maximum commits allowed for partial progress when partialProgressEnabled is true | 10 | int | | ||
| | `maxRewriteBytes(long)` | Maximum bytes to rewrite per execution | Long.MAX_VALUE | long | | ||
| | `filter(Expression)` | Filter expression for selecting files to rewrite | null | Expression | |
There was a problem hiding this comment.
Default value is Expressions.alwaysTrue()
| | `rewriteAll(boolean)` | Rewrite all data files regardless of thresholds | false | boolean | | ||
| | `maxFileGroupSizeBytes(long)` | Maximum total size of a file group | 107374182400 (100GB) | long | | ||
| | `maxFilesToRewrite(int)` | Maximum number of files to rewrite per task | Integer.MAX_VALUE | int | | ||
| | `maxFilesToRewrite(int)` | About the description we can add: If this option is not specified, all eligible files will be rewritten | null | int | |
There was a problem hiding this comment.
We don't need About the description we can add: in the doc.
Thank you for the feedback. Once I revise the parts you mentioned and add them to the Flink action docs, would that be all that’s needed? |
I don’t see any other changes needed for now. Once you finish the revisions, I will review it again. Thanks for your hard work! |
|
Thank you. I’ll reflect the points you mentioned, add them to the Flink action, and push the changes again. Thanks. |
|
How should we handle the 'Rewrite Files' action in flink-action? |
|
I have first merged the content into flink-action and deleted the original maintenance md file. Additionally, I have incorporated the two feedback points you mentioned. If there are no issues, please proceed with the merge. Thank you. |
Guosmilesmile
left a comment
There was a problem hiding this comment.
flink-configuration.mdshould not be within the scope of the current changes. Could it have been accidentally deleted?- May be we should also change the
title: "Flink Actions"in flink-actions.md totitle: "Flink TableMaintenance", and renameflink-actions.mdtoflink-maintenance.md, then change the mkdocs.yml ?
| --> | ||
|
|
||
| ## Rewrite files action | ||
| # Rewrite files action |
There was a problem hiding this comment.
| # Rewrite files action | |
| ## Flink Table Maintenance BatchMode | |
| ### Rewrite files action |
|
|
||
| For more details of the rewrite files action, please refer to [RewriteDataFilesAction](../../javadoc/{{ icebergVersion }}/org/apache/iceberg/flink/actions/RewriteDataFilesAction.html) | ||
|
|
||
| # Flink Table Maintenance |
There was a problem hiding this comment.
| # Flink Table Maintenance | |
| ## Flink Table Maintenance StreamingMode |
|
|
||
| # Flink Table Maintenance | ||
|
|
||
| ## Overview |
There was a problem hiding this comment.
| ## Overview | |
| ### Overview |
|
|
||
| The `TableMaintenance` API in **Apache Iceberg** empowers **Flink jobs** to execute maintenance tasks **natively**, either embedded within existing streaming pipelines or deployed as standalone Flink jobs. This eliminates dependencies on external systems, thereby **streamlining architecture**, **reducing operational costs**, and **enhancing automation capabilities**. | ||
|
|
||
| ## Supported Features (Flink) |
There was a problem hiding this comment.
| ## Supported Features (Flink) | |
| ### Supported Features (Flink) |
|
|
||
| ## Supported Features (Flink) | ||
|
|
||
| ### ExpireSnapshots |
There was a problem hiding this comment.
| ### ExpireSnapshots | |
| #### ExpireSnapshots |
| - Use `scheduleOnCommitCount` for write-heavy tables | ||
| - Use `scheduleOnDataFileCount` for fine-grained control | ||
|
|
||
| ### Performance Tuning |
There was a problem hiding this comment.
| ### Performance Tuning | |
| #### Performance Tuning |
| - Set reasonable `maxRewriteBytes` limits | ||
| - Setting an appropriate `maxFileGroupSizeBytes` can break down large FileGroups into smaller ones, thereby increasing the speed of parallel processing | ||
|
|
||
| ## Troubleshooting |
There was a problem hiding this comment.
| ## Troubleshooting | |
| ### Troubleshooting |
|
|
||
| ## Troubleshooting | ||
|
|
||
| ### OutOfMemoryError during file deletion |
There was a problem hiding this comment.
| ### OutOfMemoryError during file deletion | |
| #### OutOfMemoryError during file deletion |
| .deleteBatchSize(500) // Example: 500 files per batch | ||
| ``` | ||
|
|
||
| ### Lock conflicts |
There was a problem hiding this comment.
| ### Lock conflicts | |
| #### Lock conflicts |
| .rateLimit(Duration.ofMinutes(10)) // Reduce frequency of task execution | ||
| ``` | ||
|
|
||
| ### Slow rewrite operations |
There was a problem hiding this comment.
| ### Slow rewrite operations | |
| #### Slow rewrite operations |
@Guosmilesmile I’ll make the corrections you mentioned today and push the update again. Thank you for the feedback! |
|
@JeonDaehong I didn't see the relevant changes. Have you submitted them? |
Not yet! It’s currently 11 AM KST, so I should be able to start after 7 PM, about 8 hours from now. Would that be alright? Or, if it’s urgent, I can make the changes and push them even during lunch. |
|
@JeonDaehong Sorry, I thought you had already submitted the changes when I saw your reply. Just take your time. |
It seems the Korean → English translation didn’t come out quite right. Haha.. |
|
I have updated the content as per your feedback!
Please review and merge if everything looks good. Thank you! |
| ```java | ||
| import org.apache.iceberg.flink.actions.Actions; | ||
|
|
||
| TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://nn:8020/warehouse/path"); |
There was a problem hiding this comment.
fromHadoopTable -> Can we change this to hive or something else?
We would like to remove the references of the HadoopTable from the docs as it is not encouraged in production environments
|
|
||
| ```java | ||
| StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); | ||
| TableLoader tableLoader = TableLoader.fromHadoopTable("path/to/table"); |
There was a problem hiding this comment.
fromHadoopTable -> Can we change this to hive or something else?
We would like to remove the references of the HadoopTable from the docs as it is not encouraged in production environments
pvary
left a comment
There was a problem hiding this comment.
2 small changes and then we are good to go.
Thanks @JeonDaehong for the PR and @Guosmilesmile for the review!
|
Is it right to change Hadoop to Hive like this? |
| import org.apache.iceberg.flink.actions.Actions; | ||
|
|
||
| TableLoader tableLoader = TableLoader.fromCatalog( | ||
| CatalogLoader.hive("my_catalog", configuration), |
There was a problem hiding this comment.
CatalogLoader.hive("my_catalog", configuration, properties),
There was a problem hiding this comment.
I have just completed adding it.
|
Merged to main. |
|
Thank you very much for reviewing and merging! |
|
Has this Docs been visible on the Docs website since the 1.9.3 release? |
Thanks :D |
This PR adds documentation for issue
( Docs update for Flink Table Maintenance API - Fixes #13307 )
It incorporates feedback received from the previously closed PR #13360. and #13771
Please review and let me know if any changes are needed. I will update promptly.