apache / apache/beam

Add Comparator to SortValues PTransformation

Open
#21,210 0 comments 0 reactions 0 assignees View on GitHub
extensions improvement java P3 sorter
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

Code in Context:

[https://github.com/apache/beam/tree/243128a8fc52798e1b58b0cf1a271d95ee7aa241/sdks/java/extensions/sorter/src/main/java/org/apache/beam/sdk/extensions/sorter](https://github.com/apache/beam/tree/243128a8fc52798e1b58b0cf1a271d95ee7aa241/sdks/java/extensions/sorter/src/main/java/org/apache/beam/sdk/extensions/sorter)

 
Current implementation only compares serialized bytes. It would be great to have a custom comparator to sort the elements. I was able to prototype a solution but then hit some road blocks so decided to open this ticket to get some feedback.

*First approach*
We add Comparator to SortValues and propagate it down to MemorySorter, ExternalSorter. This will require adding TypeParams to all the classes including BufferedExternalSorter, MemorySorter, ExternalSorter, Sorter Interface. Instead of creating List\> we will have List\> and it will be sorted by having comparator of KV

Potential issues:
1) Since all the classes are public it does make this change not backward compatible. If we expect users to interact only with SortValues PTransform we can make this change by keeping SortValues backward compatible (If no comparator is specified we fall back to default binary comparator).
2) Both NativeExternalSorter and MemorySorter has logic to calculate memory used which can now get complicated as we are keeping deserialized objects in memory. We can mitigate around it by using `Runtime.getRuntime().freeMemory` before and after deserializing objects to estimate size of objects. (It is possible that by the time deserialization happens GC frees up some memory and we get inaccurate usage. We will have to keep a running average of the memory allocated to every record or take ratio of serialized bytes to deserialized objects)

*Second approach*:
We add Comparator to SortValues and generate Comparator out of it and use that instead. Small code snippet to show how the comparator would look like

private static class OrderingComparator implements Comparator {
private final Comparator\> comparator;
private final KvCoder, Coder\> kvCoder;

@Override
public int compare(byte[] o1, byte[] o2) {
KV kv1 = CoderUtils.decodeFromByteArray(kvCoder, o1);
KV kv2 = CoderUtils.decodeFromByteArray(kvCoder, o2);
comparator.compare(kv1, kv2);
}
}

Potential issues:
1) Sort operation is slower compared to first approach as we are serializing & deserializing objects for every comparison.
2) Memory usage: We are allocating objects in memory inside the compare method. I am not sure if they would be allocated only on the Stack (because of escape analysis) or if they would be allocated in the YoungGen. Either way they should get cleaned up quickly avoiding any memory issues.

I will create a patch after getting some feedback on this.

Imported from Jira [BEAM-12923](https://issues.apache.org/jira/browse/BEAM-12923). Original Jira may contain additional context.
Reported by: nownikhil.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.