forked from osmlab/atlas-checks
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
ShardedIntegrityChecksSparkJob Deduplication Revamp (osmlab#498)
* new dedupe * fix event service * event service per partition * debug count * debug count2 * debug count3 * part id * spotless * raw flags * concurretn queue * comments * geojson name
- Loading branch information
Showing
6 changed files
with
111 additions
and
157 deletions.
There are no files selected for viewing
This file contains 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
This file contains 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
This file contains 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
This file contains 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
117 changes: 49 additions & 68 deletions
117
src/main/java/org/openstreetmap/atlas/checks/utility/UniqueCheckFlagContainer.java
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,109 +1,90 @@ | ||
| package org.openstreetmap.atlas.checks.utility; | ||
|
|
||
| import java.io.Serializable; | ||
| import java.util.Collection; | ||
| import java.util.Collections; | ||
| import java.util.Objects; | ||
| import java.util.Set; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
| import java.util.stream.Stream; | ||
|
|
||
| import org.openstreetmap.atlas.checks.event.CheckFlagEvent; | ||
| import org.openstreetmap.atlas.checks.flag.CheckFlag; | ||
|
|
||
| /** | ||
| * A container that will deduplicate check flags based on source and unique IDs | ||
| * A container used to deduplicate check flags based on checkName and unique IDs | ||
| * | ||
| * @author jklamer | ||
| * @author bbreithaupt | ||
| */ | ||
| public class UniqueCheckFlagContainer implements Serializable | ||
| { | ||
|
|
||
| private final ConcurrentHashMap<String, ConcurrentHashMap<Set<String>, CheckFlag>> uniqueFlags; | ||
| private String checkName; | ||
| private Set<String> uniqueIdentifiers; | ||
| private CheckFlag checkFlag; | ||
|
|
||
| /** | ||
| * Combines to containers. This deduplicates {@link CheckFlag}s by overwiting ones with matching | ||
| * sources and IDs. | ||
| * | ||
| * @param container1 | ||
| * {@link UniqueCheckFlagContainer} | ||
| * @param container2 | ||
| * {@link UniqueCheckFlagContainer} | ||
| * @return merged {@link UniqueCheckFlagContainer} | ||
| * @param checkFlagEvent | ||
| * {@link CheckFlagEvent} | ||
| */ | ||
| public static UniqueCheckFlagContainer combine(final UniqueCheckFlagContainer container1, | ||
| final UniqueCheckFlagContainer container2) | ||
| public UniqueCheckFlagContainer(final CheckFlagEvent checkFlagEvent) | ||
| { | ||
| container2.uniqueFlags.entrySet() | ||
| .forEach(entry -> container1.addAll(entry.getKey(), entry.getValue().values())); | ||
| return container1; | ||
| this(checkFlagEvent.getCheckName(), checkFlagEvent.getCheckFlag().getUniqueIdentifiers(), | ||
| checkFlagEvent.getCheckFlag().makeComplete()); | ||
| } | ||
|
|
||
| public UniqueCheckFlagContainer() | ||
| /** | ||
| * @param checkName | ||
| * {@link String} check name | ||
| * @param uniqueIdentifiers | ||
| * {@link Set} of {@link String}s from {@link CheckFlag#getUniqueIdentifiers()} | ||
| * @param checkFlag | ||
| * {@link CheckFlag} | ||
| */ | ||
| public UniqueCheckFlagContainer(final String checkName, final Set<String> uniqueIdentifiers, | ||
| final CheckFlag checkFlag) | ||
| { | ||
| this.uniqueFlags = new ConcurrentHashMap<>(); | ||
| this.checkName = checkName; | ||
| this.uniqueIdentifiers = uniqueIdentifiers; | ||
| this.checkFlag = checkFlag.makeComplete(); | ||
| } | ||
|
|
||
| @SuppressWarnings("s1144") | ||
| // Ignore unused constructor warning, this is used for deserialization | ||
| private UniqueCheckFlagContainer( | ||
| final ConcurrentHashMap<String, ConcurrentHashMap<Set<String>, CheckFlag>> flags) | ||
| @Override | ||
| public boolean equals(final Object other) | ||
| { | ||
| this.uniqueFlags = flags; | ||
| if (this == other) | ||
| { | ||
| return true; | ||
| } | ||
| if (other == null || getClass() != other.getClass()) | ||
| { | ||
| return false; | ||
| } | ||
| final UniqueCheckFlagContainer that = (UniqueCheckFlagContainer) other; | ||
| return Objects.equals(this.checkName, that.checkName) | ||
| && Objects.equals(this.uniqueIdentifiers, that.uniqueIdentifiers); | ||
| } | ||
|
|
||
| /** | ||
| * Add a {@link CheckFlag} to the container based on its source. | ||
| * | ||
| * @param flagSource | ||
| * {@link String} source (check that generated the flag) | ||
| * @param flag | ||
| * {@link CheckFlag} | ||
| */ | ||
| public void add(final String flagSource, final CheckFlag flag) | ||
| public CheckFlag getCheckFlag() | ||
| { | ||
| this.uniqueFlags.putIfAbsent(flagSource, new ConcurrentHashMap<>()); | ||
| final Set<String> uniqueObjectIdentifiers = flag.getUniqueIdentifiers(); | ||
| this.uniqueFlags.get(flagSource) | ||
| .putIfAbsent(uniqueObjectIdentifiers.isEmpty() | ||
| ? Collections.singleton(flag.getIdentifier()) | ||
| : uniqueObjectIdentifiers, flag); | ||
| return this.checkFlag; | ||
| } | ||
|
|
||
| /** | ||
| * Batch add {@link CheckFlag} from a single source. | ||
| * | ||
| * @param flagSource | ||
| * {@link String} source (check that generated the flags) | ||
| * @param flags | ||
| * {@link Iterable} of {@link CheckFlag}s | ||
| */ | ||
| public void addAll(final String flagSource, final Iterable<CheckFlag> flags) | ||
| public String getCheckName() | ||
| { | ||
| flags.forEach(flag -> this.add(flagSource, flag)); | ||
| return this.checkName; | ||
| } | ||
|
|
||
| public CheckFlagEvent getEvent() | ||
| { | ||
| return new CheckFlagEvent(this.checkName, this.checkFlag); | ||
| } | ||
|
|
||
| /** | ||
| * Convert the {@link CheckFlag}s into a {@link Stream} of {@link CheckFlagEvent}s. | ||
| * | ||
| * @return a {@link Stream} of {@link CheckFlagEvent}s | ||
| */ | ||
| public Stream<CheckFlagEvent> reconstructEvents() | ||
| public Set<String> getUniqueIdentifiers() | ||
| { | ||
| return this.uniqueFlags.keySet().stream() | ||
| .flatMap(checkName -> this.uniqueFlags.get(checkName).values().stream() | ||
| .map(checkFlag -> new CheckFlagEvent(checkName, checkFlag))); | ||
| return this.uniqueIdentifiers; | ||
| } | ||
|
|
||
| /** | ||
| * Get the contents of the container as a stream. | ||
| * | ||
| * @return a {@link Stream} of {@link CheckFlag}s | ||
| */ | ||
| public Stream<CheckFlag> stream() | ||
| @Override | ||
| public int hashCode() | ||
| { | ||
| return this.uniqueFlags.values().stream().map(ConcurrentHashMap::values) | ||
| .flatMap(Collection::stream); | ||
| return Objects.hash(this.checkName, this.uniqueIdentifiers); | ||
| } | ||
| } |
Oops, something went wrong.