K1taIkuyo0421 opened a new issue, #20306: URL: https://github.com/apache/hudi/issues/20306
### Task Description ## Observed problem In a Flink MOR write benchmark using a JD-maintained Hudi 1.1 build with RemotePartitioner enabled, the sink showed approximately 7:1 input skew. Test configuration: - 10 data partitions, 10 buckets per partition, and writer parallelism 10; - bucket index keys `auction,bidder`, with repeated hot keys; - randomly generated data partition values in the range 0–9. The supplied Flink SubTasks screenshot shows: | Sink subtask | Records received | Share of total | | --- | ---: | ---: | | 0–8, each | 62,495,720–62,507,803 | approximately 6.25% | | 9 | 437,493,534 | approximately 43.75% | | All subtasks | 1,000,000,000 | 100% | Subtask 9 received approximately seven times as many records as each other subtask. The screenshot also shows approximately 118 GB received by subtask 9 versus 16.9 GB for each other subtask. <img width="539" height="475" alt="Image" src="https://github.com/user-attachments/assets/1f7d3dc2-c378-4d86-a1dc-8d8cade02197"; /> ## Cause The RemotePartitioner introduced in [#13265](https://github.com/apache/hudi/pull/13265) advances the starting offset by the bucket count for each newly assigned data partition. For fixed bucket count `b`, parallelism `c`, and bucket ID `k`, the mapping is equivalent to: ```text task = (p * b + k) % c ``` Here `p` is the server's consecutive partition assignment ordinal, starting at zero. A fixed bucket ID can reach only `c / gcd(b,c)` tasks. When `b=c=10`, the formula reduces to `task=k`: bucket 9 from every data partition goes to subtask 9. As a result, recurring same-ID hot buckets concentrate on one writer even though bucket counts are balanced. ## Suggested solution For the remote path, retain each partition's existing allocation positions but rotate the bucket positions deterministically: ```text g = gcd(b, c) shift = floor(p / (c / g)) % g task = (p * b + ((k + shift) % b)) % c ``` For fixed `b` and `c`, this preserves total bucket-count balance and makes each fixed bucket ID visit every task exactly once across any `c` consecutive assigned partitions. When `gcd(b,c)=1`, behavior is unchanged. Physical bucket IDs and record-key hashing remain unchanged. For the configuration above, this reduces to `task = (p + k) % 10`, distributing same-ID hot buckets across all ten writers while preserving the number of buckets assigned to each writer. ### Task Type Code improvement/refactoring ### Related Issues Related PR: https://github.com/apache/hudi/pull/13265 Common-factor discussion: https://github.com/apache/hudi/pull/13265#issuecomment-2862819190 -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
