Skip to content

[HUDI-6085] Eliminate cleaning tasks for flink mor table if online async compaciton is disabled #8394

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
zhuanshenbsj1 wants to merge 1 commit into from
Closed

[HUDI-6085] Eliminate cleaning tasks for flink mor table if online async compaciton is disabled #8394

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 2 additions & 0 deletions hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,8 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false);
}
return Pipelines.compact(conf, pipeline);
} else if (OptionsResolver.isMorTable(conf)) {
return Pipelines.dummySink(pipeline);
} else {

@danny0405 danny0405 Apr 6, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So what component is responsible for data cleaning then?

@zhuanshenbsj1 zhuanshenbsj1 Apr 7, 2023 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

Offline compaction do clean，Consistent with clustering.

// Append mode
  if (OptionsResolver.isAppendMode(conf)) {
    DataStream<Object> pipeline = Pipelines.append(conf, rowType, dataStream, context.isBounded());
    if (OptionsResolver.needsAsyncClustering(conf)) {
      return Pipelines.cluster(conf, rowType, pipeline);
    } else {
      return Pipelines.dummySink(pipeline);// If async-clustering=true,  no clean operator.
    }
  }

@danny0405 danny0405 Apr 9, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The cleaning can take effect finally right? Because the table would get compacted for sometime anyway.

@zhuanshenbsj1 zhuanshenbsj1 Apr 12, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Offline clustering/compaction will do clean by default , add --clean-async-enabled config will close it.

@zhuanshenbsj1 zhuanshenbsj1 Apr 13, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Current online async-clean and offline clean use the same configuration ：clean.async.enabled. Add a new configuration clean.offline.enable，making the configuration description clearer.

@danny0405 danny0405 Apr 17, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The spark offline compaction Job does not take care of cleaning, could you make it clrear how user can handle the cleaning when they use the flink streaming ingestion and spark offline compaction.

@zhuanshenbsj1 zhuanshenbsj1 Apr 19, 2023 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Adjust the cleaning operation in SparkRDDWriteClient#cluster/compact, when ASYNC_CLEAN is true will do asynchronous clean in prewrite， otherwise will do synchronous clean in autoCleanOnCommit().

@danny0405 danny0405 Apr 20, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need to think through the e2e use case for flink streaming ingestion and spark offline compaction, we should add cleaning for spark compaction and clustering first which is a block change of this patch.

@danny0405 danny0405 May 15, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We will not schedule cleaning task if async cleaning is disabled:

hudi/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/CleanFunction.java

Line 72 in 67ae0c8

if (conf.getBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED) && isCleaning) {
, so should be fine?

@zhuanshenbsj1 zhuanshenbsj1 May 15, 2023 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If CLEAN_ASYNC_ENABLED = true，a schedule will still be executed. I think cluster and compact should be consistent here(If the cow table async cluster is closed, there will be no clean operator). And now both Spark and Flink will clean up when executing offline jobs, unless forcibly closed. What do you think?

return Pipelines.clean(conf, pipeline);
}
Expand Down