[HUDI-2189] Adding delete partitions support to DeltaStreamer #4787
Conversation
| writeStatusRDD = writeClient.insertOverwriteTable(records, instantTime).getWriteStatuses(); | ||
| break; | ||
| case DELETE_PARTITION: | ||
| List<String> partitions = records.map(record -> record.getPartitionPath()).distinct().collect(); |
There was a problem hiding this comment.
If my data source is kafka, does it mean that the list of partitions to delete should be received from kafka? If answer is yes, it maybe cumbersome to combine ingestion workflow with the data and partition deletion.
There was a problem hiding this comment.
yes. thats what I have been trying to convey for a long time :) Deltastreamer (as the name suggests) is meant for incremental and continual ingestion of data from some source. It goes in cycles of fetch from source -> ingest into hudi ->repeat. So, I don't see in general how come one would fetch data from a source and then trigger delete partitions.
But we have a patch for a independent tool if you are interested. Guess that would help you. but that is a spark-submit command as well.
There was a problem hiding this comment.
My data source is Avro encoded, the schema URL can't change. It may be hard to bypass it in this case.
There was a problem hiding this comment.
ok. so, does hudi-cli work for you. I can try to add support there.
There was a problem hiding this comment.
Would be nice if it was a delta streamer CLI parameter, something that gets executed at the end of the ingestion. hudi-cli may work if it can be done as a single CLI command. For example:
hudi-cli delete-partitions --schema-name someschema --table-name sometable --location s3a://bucket/data --hive-server xxxxx --metastore=xxx --do-not-delete-data
--do-not-delete data maybe helpful to delete data faster. I use go tool that spins up hundreds of goroutines to delete hundreds of thousands of files data within seconds.
There was a problem hiding this comment.
this is against the general nomenclature of how deltastreamer works. Delta streamer consumes records from a given source and and ingests to hudi. So, I don't think we can support a cli based argument which is more of one time standalone tool/job. So, I would probably suggest you to use the standalone tool.
hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
Show resolved
Hide resolved
hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieDeltaStreamer.java
Outdated
Show resolved
Hide resolved
hudi-utilities/src/test/java/org/apache/hudi/utilities/functional/TestHoodieDeltaStreamer.java
Outdated
Show resolved
Hide resolved
hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java
Show resolved
Hide resolved
4 participants
What is the purpose of the pull request
Adding delete partition support to deltastreamer.
Brief change log
(for example:)
Verify this pull request
(Please pick either of the following options)
This pull request is a trivial rework / code cleanup without any test coverage.
(or)
This pull request is already covered by existing tests, such as (please describe tests).
(or)
This change added tests and can be verified as follows:
(example:)
Committer checklist
Has a corresponding JIRA in PR title & commit
Commit message is descriptive of the change
CI is green
Necessary doc changes done or have another open PR
For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.