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
mpol
approved these changes
Jul 10, 2026
mpol
left a comment
mpol
left a comment
Contributor
There was a problem hiding this comment.
Looks good. Only some comments on comments.
Comment on lines
+47
to
+55
| /** | ||
| * Number of random salt values used to split hot buckets. Affiliations in a hot bucket are | ||
| * assigned a random salt in [0, SALT_FACTOR), while organizations are replicated with every | ||
| * salt value, so the cross-product is distributed across SALT_FACTOR independent tasks. | ||
| * <p> | ||
| * With the observed ~280x skew ratio between the hottest task and the p75, a factor of 20 | ||
| * is sufficient to bring the hottest task close to the cluster median. | ||
| */ | ||
| static final int SALT_FACTOR = 20; |
Contributor
There was a problem hiding this comment.
If I understand correctly, a salt factor of 20 should reduce the largest buckets about 20 times. If so and with skew ratio of ~280 between it and p75 this seems still pretty far from the median. Possibly this is a good compromise value anyway, especially as the organizations get replicated, but the "close to the median" comment seems misleading.
Comment on lines
+22
to
+30
| /** | ||
| * Maximum number of organization associations allowed per document when building the join key. | ||
| * Documents linked to more organizations than this threshold (e.g. a paper with many grant partners) | ||
| * are the primary cause of data skew in CoGroupedRDD [43]: one task handles the entire Cartesian | ||
| * product of (affiliations × organizations) for a hot document ID, producing partitions orders of | ||
| * magnitude larger than the median. Such documents contribute very little signal for affiliation | ||
| * matching because the affiliation–organization pairing becomes essentially random at this scale. | ||
| * Tune this value based on observed skew; 100 is a conservative starting point. | ||
| */ |
Contributor
There was a problem hiding this comment.
This comment seems to be referring to a particular stage in the particular plan, so could be made more generic.
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.