[flink] Flink sink support hash by bucket id for PrimaryKey Table #579
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
swuferhong
force-pushed
the
hash-by-bucket-key
branch
from
69a01ee to
cb28ec4
Compare
wuchong
changed the title
Member
|
The test coverage is failed: |
wuchong
requested changes
Mar 12, 2025
...s-connector-flink/src/main/java/com/alibaba/fluss/connector/flink/FlinkConnectorOptions.java
Outdated
Show resolved
Hide resolved
...s-connector-flink/src/main/java/com/alibaba/fluss/connector/flink/FlinkConnectorOptions.java
Outdated
Show resolved
Hide resolved
...rs/fluss-connector-flink/src/main/java/com/alibaba/fluss/connector/flink/sink/FlinkSink.java
Outdated
Show resolved
Hide resolved
...rs/fluss-connector-flink/src/main/java/com/alibaba/fluss/connector/flink/sink/FlinkSink.java
Outdated
Show resolved
Hide resolved
...connector-flink/src/main/java/com/alibaba/fluss/connector/flink/sink/RowDataKeySelector.java
Outdated
Show resolved
Hide resolved
...uss-connector-flink/src/main/java/com/alibaba/fluss/connector/flink/sink/FlinkTableSink.java
Outdated
Show resolved
Hide resolved
fluss-common/src/main/java/com/alibaba/fluss/bucketing/BucketingFunction.java
Outdated
Show resolved
Hide resolved
2 tasks
swuferhong
force-pushed
the
hash-by-bucket-key
branch
6 times, most recently
from
9b95593 to
de907f9
Compare
Contributor
Author
|
@wuchong comments addressed. |
2 tasks
wuchong
reviewed
Mar 24, 2025
...uss-flink-common/src/main/java/com/alibaba/fluss/flink/sink/FlinkRowDataChannelComputer.java
Outdated
Show resolved
Hide resolved
...uss-flink-common/src/main/java/com/alibaba/fluss/flink/sink/FlinkRowDataChannelComputer.java
Show resolved
Hide resolved
wuchong
reviewed
Mar 24, 2025
Comment on lines
+160
to
+170
| void testAppendLogWithBucketKeyWithSinkBucketShuffle() throws Exception { | ||
| testAppendLogWithBucketKey(true); | ||
| } | ||
|
|
||
| @Test | ||
| void testAppendLogWithBucketKeyWithoutSinkBucketShuffle() throws Exception { | ||
| testAppendLogWithBucketKey(false); | ||
| } | ||
|
|
||
| private void testAppendLogWithBucketKey(boolean sinkBucketShuffle) throws Exception { |
Member
There was a problem hiding this comment.
Did you try @ParameterizedTest? It would be better to use @ParameterizedTest if possible.
Contributor
Author
There was a problem hiding this comment.
I tried.... But I encountered the same problem as when I developed #351 (add parameterized test in FlinkTableSinkITCase) : When using @ParameterizedTest, the CI on GitHub will hang to timeout, but in local env tests will passed. I'll create an issue to trace this problem and look into the root cause later: #659
Member
There was a problem hiding this comment.
Thanks, I think we need to investigate the reason, otherwise, we don't know whether to add @ParameterizedTest or not.
swuferhong
force-pushed
the
hash-by-bucket-key
branch
from
de907f9 to
21c27b3
Compare
wuchong
approved these changes
Mar 24, 2025
ZmmBigdata
pushed a commit
to ZmmBigdata/fluss
that referenced
this pull request
Jun 20, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Purpose
Linked issue: #570
This pr is aims to support hash by bucket key for
PrimaryKeyTable for Flink sink. The approach of this pr is to first hash the data bybucket idbefore writing to the Flink sink, sending data with the samebucket idto the same sink parallelism. This allows the Fluss writer to batch kv data more effectively, thus reducing the pressure on the client and improving throughput. From the testing results, this can greatly enhance the throughput of writing FlussPrimaryKeytables and reduce memory usage, while also reducing pressure on the server side (fewerputKvrequests and larger CDC log batches).Brief change log
Tests
API and Format
Documentation