You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2022/03/02 13:46:26 UTC
[GitHub] [hudi] xushiyan commented on a change in pull request #4856: [HUDI-2439] Replace RDD with HoodieData in HoodieSparkTable and commit executors
xushiyan commented on a change in pull request #4856:
URL: https://github.com/apache/hudi/pull/4856#discussion_r817704373
##########
File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/HoodieDeleteHelper.java
##########
@@ -48,69 +45,62 @@
* @param <T>
*/
@SuppressWarnings("checkstyle:LineLength")
-public class SparkDeleteHelper<T extends HoodieRecordPayload,R> extends
- BaseDeleteHelper<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>, R> {
- private SparkDeleteHelper() {
+public class HoodieDeleteHelper<T extends HoodieRecordPayload, R> extends
+ BaseDeleteHelper<T, HoodieData<HoodieRecord<T>>, HoodieData<HoodieKey>, HoodieData<WriteStatus>, R> {
+ private HoodieDeleteHelper() {
}
private static class DeleteHelperHolder {
- private static final SparkDeleteHelper SPARK_DELETE_HELPER = new SparkDeleteHelper();
+ private static final HoodieDeleteHelper HOODIE_DELETE_HELPER = new HoodieDeleteHelper<>();
}
- public static SparkDeleteHelper newInstance() {
- return DeleteHelperHolder.SPARK_DELETE_HELPER;
+ public static HoodieDeleteHelper newInstance() {
+ return DeleteHelperHolder.HOODIE_DELETE_HELPER;
}
@Override
- public JavaRDD<HoodieKey> deduplicateKeys(JavaRDD<HoodieKey> keys, HoodieTable<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>> table, int parallelism) {
+ public HoodieData<HoodieKey> deduplicateKeys(HoodieData<HoodieKey> keys, HoodieTable<T, HoodieData<HoodieRecord<T>>, HoodieData<HoodieKey>, HoodieData<WriteStatus>> table, int parallelism) {
boolean isIndexingGlobal = table.getIndex().isGlobal();
if (isIndexingGlobal) {
- return keys.keyBy(HoodieKey::getRecordKey)
- .reduceByKey((key1, key2) -> key1, parallelism)
- .values();
+ return keys.distinctWithKey(HoodieKey::getRecordKey, parallelism);
} else {
- return keys.distinct(parallelism);
+ return keys.distinct();
}
}
@Override
- public HoodieWriteMetadata<JavaRDD<WriteStatus>> execute(String instantTime,
- JavaRDD<HoodieKey> keys,
- HoodieEngineContext context,
- HoodieWriteConfig config,
- HoodieTable<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>> table,
- BaseCommitActionExecutor<T, JavaRDD<HoodieRecord<T>>, JavaRDD<HoodieKey>, JavaRDD<WriteStatus>, R> deleteExecutor) {
- JavaSparkContext jsc = HoodieSparkEngineContext.getSparkContext(context);
-
+ public HoodieWriteMetadata<HoodieData<WriteStatus>> execute(String instantTime,
+ HoodieData<HoodieKey> keys,
+ HoodieEngineContext context,
+ HoodieWriteConfig config,
+ HoodieTable<T, HoodieData<HoodieRecord<T>>, HoodieData<HoodieKey>, HoodieData<WriteStatus>> table,
+ BaseCommitActionExecutor<T, HoodieData<HoodieRecord<T>>, HoodieData<HoodieKey>, HoodieData<WriteStatus>, R> deleteExecutor) {
try {
- HoodieWriteMetadata result = null;
- JavaRDD<HoodieKey> dedupedKeys = keys;
+ HoodieData<HoodieKey> dedupedKeys = keys;
final int parallelism = config.getDeleteShuffleParallelism();
if (config.shouldCombineBeforeDelete()) {
// De-dupe/merge if needed
dedupedKeys = deduplicateKeys(keys, table, parallelism);
- } else if (!keys.partitions().isEmpty()) {
- dedupedKeys = keys.repartition(parallelism);
Review comment:
this was spark-specific logic; need to discuss the implication
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org