[HUDI-759] Integrate checkpoint provider with delta streamer #1486

garyli1019 · 2020-04-05T05:25:15Z

What is the purpose of the pull request

Integrate the initial checkpoint provider with delta streamer

Brief change log

Add two options to delta streamer to use the initial checkpoint provider

Verify this pull request

This change added tests and can be verified as follows:

Added unit test in TestHoodieDeltaStreamer

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.

garyli1019 · 2020-04-05T05:27:58Z

@pratyakshsharma @vinothchandar
I agree with Vinoth's idea that having those two options in the delta streamer. Is this implementation makes sense to you guys?
I will add a test case if this approach looks good.
Thanks

codecov-io · 2020-04-05T05:58:39Z

Codecov Report

Merging #1486 into master will decrease coverage by 0.07%.
The diff coverage is 78.78%.

@@             Coverage Diff              @@
##             master    #1486      +/-   ##
============================================
- Coverage     72.23%   72.15%   -0.08%     
- Complexity      289      294       +5     
============================================
  Files           338      373      +35     
  Lines         15947    16282     +335     
  Branches       1624     1638      +14     
============================================
+ Hits          11519    11748     +229     
- Misses         3700     3798      +98     
- Partials        728      736       +8

Impacted Files	Coverage Δ	Complexity Δ
...in/java/org/apache/hudi/utilities/UtilHelpers.java	`64.70% <50.00%> (-0.71%)`	`22.00 <1.00> (+1.00)`	⬇️
...lities/checkpointing/KafkaConnectHdfsProvider.java	`89.28% <71.42%> (-3.03%)`	`14.00 <3.00> (+2.00)`	⬇️
...ities/checkpointing/InitialCheckPointProvider.java	`83.33% <83.33%> (ø)`	`1.00 <1.00> (?)`
...i/utilities/deltastreamer/HoodieDeltaStreamer.java	`77.93% <85.71%> (-1.22%)`	`11.00 <4.00> (+1.00)`	⬇️
...apache/hudi/utilities/deltastreamer/DeltaSync.java	`72.44% <100.00%> (ø)`	`37.00 <0.00> (ø)`
...g/apache/hudi/metrics/InMemoryMetricsReporter.java	`40.00% <0.00%> (-40.00%)`	`0.00% <0.00%> (ø%)`
.../org/apache/hudi/table/HoodieCopyOnWriteTable.java	`61.62% <0.00%> (-27.66%)`	`0.00% <0.00%> (ø%)`
.../org/apache/hudi/table/HoodieMergeOnReadTable.java	`57.50% <0.00%> (-25.63%)`	`0.00% <0.00%> (ø%)`
...hudi/common/fs/inline/InLineFsDataInputStream.java	`38.46% <0.00%> (-15.39%)`	`0.00% <0.00%> (ø%)`
...src/main/java/org/apache/hudi/metrics/Metrics.java	`58.33% <0.00%> (-13.89%)`	`0.00% <0.00%> (ø%)`
... and 51 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 5d717a2...b923a97. Read the comment docs.

pratyakshsharma · 2020-04-05T20:52:21Z

LGTM

garyli1019 · 2020-04-05T22:12:20Z

Test added. Thanks for the review

garyli1019 · 2020-04-08T01:31:38Z

Add #1493 into this PR.

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java

vinothchandar

Few clarifications

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java

vinothchandar · 2020-04-09T05:35:45Z

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java

+    if (cfg.initialCheckpointProvider != null && cfg.bootstrapFromPath != null && cfg.checkpoint == null) {
+      InitialCheckPointProvider checkPointProvider =
+          UtilHelpers.createInitialCheckpointProvider(cfg.initialCheckpointProvider, new Path(cfg.bootstrapFromPath), fs);
+      cfg.checkpoint = checkPointProvider.getCheckpoint();


IIUC setting cfg.checkpoint will force use of that timestamp instead of what we normally do - read from the last commit?

Should we do this only when creating the dataset for the first time..

I think this depends on how we design the migration flow for the user.
What I did myself is I use Spark datasource to do a bulkInsert to convert all the plain parquet files to Hudi format, then the second job I'd like to use delta streamer to read from Kafka. So this initialCheckpointProvider should be the first delta streamer job when switching sources.

yes.. do you think if we made it such that even if someone runs delta streamer few times after initial bootstrap, the initial checkpoint provider is used just once? otherwise, you need to scramble to stop the delta streamer after the first run or manually run it by hand once before scheduling it using airflow or deploying in --continuous mode?

I believe the initial checkpoint provider should be just used once when the user wants to switch from one source to another. After that, the delta streamer should be able to get the checkpoint from the previous commit. We can improve this once the bootstrap is ready. At this point, I am not sure how to put everything together if we want one step to handling everything.

okay.. lets revisit once we have bootstrap support.. cc @bvaradar as fyi

hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieDeltaStreamer.java

vinothchandar

Left some suggestions.. lmk what you think

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java

...ilities/src/main/java/org/apache/hudi/utilities/checkpointing/InitialCheckPointProvider.java

garyli1019 · 2020-04-10T05:59:52Z

hmm... Looks like the checkstyle auto-fix something...Let me see what's going...

garyli1019 · 2020-04-10T06:20:09Z

I added the save action and checkstyle as documented, not sure which one triggered all those final and this. Is that ok to add those? As a scala programmer I do prefer to use val all the time and the Google Java Guide does encourage using final, but I am not sure what is the preference of Hudi.

garyli1019 · 2020-04-10T18:58:47Z

Addressed some comments, summary:

Removed --bootstrap-from option in the delta streamer. Use hoodie.deltastreamer.checkpoint.provider.path field in the props instead.
Use TypedProperty to construct the InitialCheckPointProvider and init(FileSystem fs) to initialize the class
Keep hiveConf as the variable name even change HiveConf type to Configuration. Open to discussion if you guys don't agree.
Not able to replace all null in this PR because null was served as a flag in the delta streamer workflow, this might change the behavior of other classes using the TypedProperty field. Will need a separate PR to do the code refactoring.
The style check tool automatically adds final and this to match the stylecheck.xml.

vinothchandar

Please revert the non-essential fixes from the PR (final, this..) and so on., so its easier to review.. Those need more discussion if we are changing code style and applied uniformly..

hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java

vinothchandar

The style check tool automatically adds final and this to match the stylecheck.xml.

Please revert this change..

vinothchandar

Thanks for working through this @garyli1019 .. Hopefully just one more cycle.. and we are home.

hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java

...ilities/src/main/java/org/apache/hudi/utilities/checkpointing/InitialCheckPointProvider.java

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java

vinothchandar · 2020-04-13T00:53:04Z

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java

+    if (cfg.initialCheckpointProvider != null && cfg.bootstrapFromPath != null && cfg.checkpoint == null) {
+      InitialCheckPointProvider checkPointProvider =
+          UtilHelpers.createInitialCheckpointProvider(cfg.initialCheckpointProvider, new Path(cfg.bootstrapFromPath), fs);
+      cfg.checkpoint = checkPointProvider.getCheckpoint();


yes.. do you think if we made it such that even if someone runs delta streamer few times after initial bootstrap, the initial checkpoint provider is used just once? otherwise, you need to scramble to stop the delta streamer after the first run or manually run it by hand once before scheduling it using airflow or deploying in --continuous mode?

garyli1019 · 2020-04-14T00:33:03Z

@vinothchandar Thanks for all the feedback! Very helpful!
Comments addressed. Please take a look.

vinothchandar · 2020-04-14T05:01:14Z

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java

+    @Parameter(names = {"--initial-checkpoint-provider"}, description = "Generate check point for delta streamer "
+        + "for the first run. This field will override the checkpoint of last commit using the checkpoint field. "
+        + "Use this field only when switch source, for example, from DFS source to Kafka Source. Check the class "
+        + "org.apache.hudi.utilities.checkpointing for details")


InitialCheckPointProvider did you intend to write the name of the class here?

Do you mean we should add InitialCheckPointProvider here or we should remove this description?

Changed the description to match with --schemaprovider-class

vinothchandar

LGTM.. 1 minor comment. once you respond/repush.. can merge

garyli1019 force-pushed the HUDI-759 branch from 5329fc8 to c904218 Compare April 5, 2020 22:09

garyli1019 changed the title ~~WIP[HUDI-759] Integrate checkpoint privoder with delta streamer~~ [HUDI-759] Integrate checkpoint privoder with delta streamer Apr 5, 2020

vinothchandar self-assigned this Apr 6, 2020

garyli1019 force-pushed the HUDI-759 branch 2 times, most recently from 12957f3 to 0304288 Compare April 8, 2020 00:37

garyli1019 mentioned this pull request Apr 8, 2020

[MINOR] remove Hive dependency from delta streamer #1493

Closed

5 tasks

garyli1019 force-pushed the HUDI-759 branch from 0304288 to 41f78ec Compare April 8, 2020 01:29

pratyakshsharma reviewed Apr 8, 2020

View reviewed changes

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java Outdated Show resolved Hide resolved

pratyakshsharma reviewed Apr 8, 2020

View reviewed changes

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java Outdated Show resolved Hide resolved

pratyakshsharma reviewed Apr 8, 2020

View reviewed changes

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java Outdated Show resolved Hide resolved

pratyakshsharma reviewed Apr 8, 2020

View reviewed changes

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java Outdated Show resolved Hide resolved

vinothchandar reviewed Apr 9, 2020

View reviewed changes

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java Show resolved Hide resolved

vinothchandar reviewed Apr 9, 2020

View reviewed changes

hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieDeltaStreamer.java Outdated Show resolved Hide resolved

vinothchandar reviewed Apr 9, 2020

View reviewed changes

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java Outdated Show resolved Hide resolved

...ilities/src/main/java/org/apache/hudi/utilities/checkpointing/InitialCheckPointProvider.java Outdated Show resolved Hide resolved

garyli1019 force-pushed the HUDI-759 branch from 41f78ec to 2e4ebda Compare April 10, 2020 05:49

garyli1019 force-pushed the HUDI-759 branch from 2e4ebda to 639b972 Compare April 10, 2020 18:48

vinothchandar changed the title ~~[HUDI-759] Integrate checkpoint privoder with delta streamer~~ [HUDI-759] Integrate checkpoint provider with delta streamer Apr 12, 2020

vinothchandar reviewed Apr 12, 2020

View reviewed changes

hudi-utilities/src/main/java/org/apache/hudi/utilities/UtilHelpers.java Outdated Show resolved Hide resolved

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java Outdated Show resolved Hide resolved

vinothchandar requested changes Apr 12, 2020

View reviewed changes

garyli1019 force-pushed the HUDI-759 branch 2 times, most recently from 92cf0e7 to 94bb6ff Compare April 12, 2020 19:25

garyli1019 force-pushed the HUDI-759 branch 2 times, most recently from a660f04 to 51740fc Compare April 12, 2020 19:51

garyli1019 requested a review from vinothchandar April 12, 2020 20:15

vinothchandar reviewed Apr 13, 2020

View reviewed changes

garyli1019 force-pushed the HUDI-759 branch from 51740fc to 0241380 Compare April 14, 2020 00:28

vinothchandar reviewed Apr 14, 2020

View reviewed changes

vinothchandar approved these changes Apr 14, 2020

View reviewed changes

garyli1019 force-pushed the HUDI-759 branch from 0241380 to d4d3768 Compare April 14, 2020 05:23

HUDI-759 Integrate checkpoint provider with delta streamer

b923a97

garyli1019 force-pushed the HUDI-759 branch from d4d3768 to b923a97 Compare April 14, 2020 05:24

vinothchandar merged commit 14d4fea into apache:master Apr 14, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[HUDI-759] Integrate checkpoint provider with delta streamer #1486

[HUDI-759] Integrate checkpoint provider with delta streamer #1486

garyli1019 commented Apr 5, 2020 •

edited

garyli1019 commented Apr 5, 2020

codecov-io commented Apr 5, 2020 •

edited

pratyakshsharma commented Apr 5, 2020

garyli1019 commented Apr 5, 2020

garyli1019 commented Apr 8, 2020

vinothchandar left a comment

vinothchandar Apr 9, 2020

garyli1019 Apr 10, 2020

vinothchandar Apr 13, 2020

garyli1019 Apr 14, 2020

vinothchandar Apr 14, 2020

vinothchandar left a comment

garyli1019 commented Apr 10, 2020

garyli1019 commented Apr 10, 2020

garyli1019 commented Apr 10, 2020

vinothchandar left a comment

vinothchandar left a comment •

edited

vinothchandar left a comment

vinothchandar Apr 13, 2020

garyli1019 commented Apr 14, 2020

vinothchandar Apr 14, 2020

garyli1019 Apr 14, 2020

garyli1019 Apr 14, 2020

vinothchandar left a comment

[HUDI-759] Integrate checkpoint provider with delta streamer #1486

[HUDI-759] Integrate checkpoint provider with delta streamer #1486

Conversation

garyli1019 commented Apr 5, 2020 • edited

What is the purpose of the pull request

Brief change log

Verify this pull request

Committer checklist

garyli1019 commented Apr 5, 2020

codecov-io commented Apr 5, 2020 • edited

Codecov Report

pratyakshsharma commented Apr 5, 2020

garyli1019 commented Apr 5, 2020

garyli1019 commented Apr 8, 2020

vinothchandar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vinothchandar left a comment

Choose a reason for hiding this comment

garyli1019 commented Apr 10, 2020

garyli1019 commented Apr 10, 2020

garyli1019 commented Apr 10, 2020

vinothchandar left a comment

Choose a reason for hiding this comment

vinothchandar left a comment • edited

Choose a reason for hiding this comment

vinothchandar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

garyli1019 commented Apr 14, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vinothchandar left a comment

Choose a reason for hiding this comment

garyli1019 commented Apr 5, 2020 •

edited

codecov-io commented Apr 5, 2020 •

edited

vinothchandar left a comment •

edited