Marekhorst 1611 optimize affiliation matching performance by addressing potential data skew#1612
Open
marekhorst wants to merge 4 commits into
Conversation
… potential data skew Introducing initial commit. Optimize affiliation matching Spark job to fix 17h runtime in Stage 16 Stage 16 suffered from two compounding problems: * a single giant stage spanning all 5 matchers (80 RDDs, 101 GB shuffle spill, 13.6 GB disk spill) * and severe data skew causing one task to run 16.3h vs p75 of 0.9s. Changes: * `AffMatchingService`: caching `normalizedAffiliations` and `enrichedOrganizations` so all 5 matchers share a single read+normalize pass instead of recomputing from source independently * `AffMatchingService`: add per-matcher `reduceByKey()` before the union to force a shuffle boundary after each matcher, splitting the single Stage 16 into 6 smaller independently-sized and retryable stages; also reduces data volume fed into the final cross-matcher dedup * `DocOrgRelationAffOrgJoiner`: filter out documents with more than 100 organization associations before the `documentId` join to eliminate the skewed hot-key partition that caused the 16.3h outlier task * `workflow.xml`: add `spark.default.parallelism=2560` (2x cluster vcores) and tune `spark.memory.fraction/storageFraction` to reduce shuffle spill
… potential data skew Fix data skew in AffOrgHashBucketJoiner via salted join Hash bucket matchers suffered severe skew because coarse hash functions map all "University of X" names to a single bucket: StringPartFirstLetters Hasher(2 parts, 2 letters) produces "UnOf" for every university affiliation, and OrganizationSectionHasher always produces the same hash for "University of X" names where wordBefore="" and wordAfter="of". This caused one task to run 16h in Stage 20 and 10h in Stage 19 while p75 was 49s and 22s. Replace the plain join in AffOrgHashBucketJoiner with a salted join: - Count affiliation and organization entries per hash bucket - Collect hot keys (cross-product > 1 000 000) to the driver and broadcast - For hot buckets: assign a random salt in [0, 20) to each affiliation and replicate each organization with all 20 salt values, distributing the Cartesian product across 20 independent tasks instead of one - For cold buckets: use the original join unchanged (fast path) No matches are lost: every affiliation with salt N meets every organization at salt N. The underlying affiliation/organization RDDs are cached so the bucket-counting passes reuse memory without re-reading from disk.
… potential data skew Aligning tests with the optimization updates. Updating comments to be more appropriate by being less specific to a particular case (address specific stage numbers etc).
Contributor
There was a problem hiding this comment.
Pull request overview
This PR optimizes the affiliation→organization matching Spark workflow to reduce extreme data skew and overall runtime, primarily by (1) salting “hot” hash buckets to spread large Cartesian products across tasks, and (2) reducing shuffle volume / stage size via caching and earlier deduplication.
Changes:
- Add hot-bucket detection + salted join path to distribute oversized hash-bucket Cartesian products (
AffOrgHashBucketJoiner), with a fast-path when no hot buckets exist. - Reduce per-matcher and cross-matcher shuffle pressure by caching shared RDDs and adding per-matcher
reduceByKeydeduplication (AffMatchingService+ updated unit test). - Add skew-prevention filtering to the doc-org joiner and tune Spark workflow runtime parameters (
DocOrgRelationAffOrgJoiner,workflow.xml), plus a small voter strength adjustment.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| iis-wf/iis-wf-affmatching/src/main/java/eu/dnetlib/iis/wf/affmatching/bucket/AffOrgHashBucketJoiner.java | Detects hot hash buckets and performs salted joins to mitigate skew; keeps a plain-join fast path. |
| iis-wf/iis-wf-affmatching/src/test/java/eu/dnetlib/iis/wf/affmatching/bucket/AffOrgHashBucketJoinerTest.java | Updates unit test to cover the cold/no-hot-bucket fast path and the new hot-key detection chain. |
| iis-wf/iis-wf-affmatching/src/main/java/eu/dnetlib/iis/wf/affmatching/AffMatchingService.java | Caches shared RDDs and adds per-matcher deduplication to reduce stage size and shuffle volume. |
| iis-wf/iis-wf-affmatching/src/test/java/eu/dnetlib/iis/wf/affmatching/AffMatchingServiceTest.java | Adapts expectations to the new per-matcher dedup + union + final dedup pipeline. |
| iis-wf/iis-wf-affmatching/src/main/java/eu/dnetlib/iis/wf/affmatching/bucket/DocOrgRelationAffOrgJoiner.java | Filters out high-fanout document→org keys to reduce doc-based join skew. |
| iis-wf/iis-wf-affmatching/src/main/resources/eu/dnetlib/iis/wf/affmatching/main/oozie_app/workflow.xml | Adds Spark runtime tuning (default parallelism, memory fractions) for the Oozie Spark action. |
| iis-wf/iis-wf-affmatching/src/main/java/eu/dnetlib/iis/wf/affmatching/match/DocOrgRelationMatcherFactory.java | Adjusts commonOrgNameWordsVoter match strength to align with the updated pipeline. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
54
to
+61
| JavaRDD<AffMatchDocumentOrganization> documentOrganizations = documentOrganizationFetcher.fetchDocumentOrganizations(); | ||
| JavaPairRDD<String, AffMatchDocumentOrganization> documentOrganizationDocIdKey = documentOrganizations.keyBy(docOrg -> docOrg.getDocumentId()); | ||
|
|
||
| // Count organizations per document to identify skewed (high-fanout) keys. | ||
| // Documents linked to an excessive number of organizations via project chains create | ||
| // Cartesian-product partitions in the join below that dwarf all other partitions combined. | ||
| JavaPairRDD<String, Long> orgCountPerDoc = documentOrganizations | ||
| .mapToPair(docOrg -> new Tuple2<>(docOrg.getDocumentId(), 1L)) | ||
| .reduceByKey(Long::sum); |
Comment on lines
+141
to
+143
| // Start with an empty typed RDD rather than parallelizePairs on an empty list to avoid | ||
| // carrying an empty RDD through all union() calls | ||
| JavaPairRDD<Tuple2<String, String>, AffMatchResult> allMatchedAffOrgsWithKey = null; |
Comment on lines
+95
to
+101
| Set<String> hotKeySet = affCounts | ||
| .join(orgCounts) | ||
| .filter(x -> (double) x._2._1 * x._2._2 > MAX_BUCKET_CROSS_PRODUCT) | ||
| .keys() | ||
| .collect() | ||
| .stream() | ||
| .collect(Collectors.toSet()); |
Member
Author
There was a problem hiding this comment.
After running a test on production data and obtaining all the hot keys it doesn't seem like we need to be worried about exceeding this 8GB broadcast limit. See #1611 (comment) for more details.
… potential data skew Addresing the first round of code-review comments: * introducing caching of documentOrganizations RDD which is consumed twice * updating a comment which is considered as misleading
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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
This PR introduces a set of optimizations aiming to address the significant data skew and very long (17h) execution time.
Apart from the bunch of generic optimizations from the first commit (which were needed but did not affect the execution time in a significant way) the second commit introduces salting for extremely large aff*org Cartesian products. This is the most important change affecting both the data skew and, as a result of that, execution time which was reduced to 2-3 hours.
Due to the optimization introduced in the first commit the
DocOrgRelationMatcherFactorythe match strength of thecommonOrgNameWordsVoterhad to be updated.