[BEAM-6812]: Convert keys to ByteArray in Combine.perKey to make sure hashCode is consistent

[jhalaria]

…
The PR makes the CBK consistent with GBK where we convert the keys to a ByteArray before calling group/combine functions.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Flink	Gearpump	Samza	Spark
Go		---	---	---	---	---	---
Java
Python		---			---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

[jhalaria]

@iemejia - Please review.

[jhalaria]

This logic seems to be accidentally flipped?

[dmvk]

agreed, this logic is odd, I don't think EvalutionContext is enough to properly resolve paralellism for the shuffle operation

In case bundleSize > 0, we should reuse paralellism of the input RDD. This way we would get rid of nullable case and we can simplify GBK and CPK translations.

Spark's default partitioner does the same thing: org.apache.spark.Partitioner#defaultPartitioner

[jhalaria]

@iemejia : Please take as lot this when you get a chance. Thanks.

[iemejia]

Sorry for the delay, have not forgotten about this one, just being swamped in other PRs. @dmvk could you help me review this one please.

[dmvk]

Run Portable_Python PreCommit

[dmvk]

Nice catch! This problem goes way deeper, we should not ever shuffle raw user data in the first place (we should always use beam coder to serialize data, before partitioning even happens). If you take a look at GBK implementation, it uses ByteArray instead of byte[] (which has correct hashCode implementation).

I guess tests did not catch this because they use primitive types (which spark can serialize on its own) as keys. Also tests are run in a single JVM, therefore hashCode is stable.

Correct fix would be:

• Changing CombinePerKey implementation to convert keys to ByteArray prior calling combineByKey
• In case we want to use custom partitioner, we should always check on the first getPartition call, whether hashCode == systemIdentityHashCode and throw an exception if it does; or make sure that we shuflle byte[] only (which may be less expensive than ByteArray object)

[dmvk]

This is a great start! It would be awesome if we could cover all cases by properly using Beam's coders for all shuffle operations as expected.

Thanks you for the contribution! ;)

[jhalaria]

@dmvk - Thank you for looking at the PR. I removed the custom partitioner logic for now. I made the CPK similar to GBK where we convert the key to a ByteArray. Will create another PR that provides the ability to add a custom partitioner.

[dmvk]

I think we should rather keep the getPartition mechanism the way it was for now. Otherwise it is good to merge! Thanks ;)

[dmvk]

Second thoughts, it was correct.

https://github.com/apache/beam/pull/6884/files#r246077919

This was in order to maintain old functionality (bundleSize == 0, which basically means to use predefined parallelism).

I think the old functionality doesn't make much sense as it doesn't scale with input data. I guess someone may use this in order to re-scale "downstream" stage, but there should be a better mechanism to do this.

Any thoughts? @timrobertson100 @kyle-winkelman

[jhalaria]

I see. I reverted back the changes made to getPartition

[kyle-winkelman]

I agree that the old functionality seems strange, but I remember (when I had the logic backwards) that the performance tests for the spark runner were impacted. I think the impact was in streaming mode because if you don't use this HashPartitioner then it actually does a double shuffle of the data. I tried to clean this up but my I never finished getting PR #6511 merged.

[dmvk]

Run Spark ValidatesRunner