[NEMO-472] Implement Intermediate Combine

[Kangji]

JIRA: NEMO-472: Fix and Implement Hierarchical Aggregation

Major changes:
[NEMO-472: Implement Hierarchical Aggregation] aims to add additional intermediate accumulation operator in front of final combine operator that accumulates data among physically nearby containers prior to shuffling across WAN, when needed. It is expected that data aggregation among nearby containers will reduce the data size that must be transferred across WAN. To achieve it,

• Implemented intermediate combine transform
  • Previous Combine.PerKey Transform consisted of 2 steps.
    1. Partial Combine(a.k.a. pre-aggregation): accumulates elements in each containers. Therefore, data transfer across network is not needed in this step.
    2. Final Combine: shuffle all data(hashed by key) and then combine.
  • Additional, and optional step that accumulates the pre-aggregated data partially(only among nearby containers) is implemented and inserted between 1(partial) and 2(final).
  • This new type of transform is only used in intermediate accumulator vertex, which is special type of operator vertex.
• Added new type of communication channel, Partial Shuffle, which represents data transfer from upstream operator to intermediate accumulator vertex. It resembles shuffle, but the difference is that data shuffle occurs only among physically nearby containers.
• Implemented compile time optimization pass that inserts intermediate accumulator vertex, which performs hierarchical aggregation prior to shuffle, only when it is expected to be effective.
• Implemented unit tests.

Minor changes to note:

• None

Tests for the changes:

• Tested on my Mac and ubuntu machine

Other comments:

• Data transfer on partial shuffle communication channel is implemented in [Nemo-472] Implement Runtime Components of Hierarchical Aggregation #319.
• [TODO] Need more conditions to be implemented to make decision whether applying the pass is effective or not. Current logic is too naive.

[taegeonum]

I did my first pass. Overall, looks good to me. Thanks for the work @Kangji !

[taegeonum]

Could you explain why the class name is changed?

[Kangji]

It's because GroupByKey Transform in Beam semantics can be represented as Combine PerKey Transform, and I thought that this class represents Combine PerKey rather than GroupByKey.

[taegeonum]

As far as I know, GroupByKey Transform is not always be represented as Combine PerKey Transform, so changing the name is confusing to me. For instance, CoGroupByKey is not combining, but it is also represented as GroupByKey as far as I know.

[wonook]

We actually have a separate GroupByKeyTransform. We have renamed the GBKTransform since it actually works as a CombineTransform. It was misnamed in the first place.

[sonarqubecloud]

Kudos, SonarCloud Quality Gate passed!   

0 Bugs
0 Vulnerabilities
0 Security Hotspots
0 Code Smells

No Coverage information
0.0% Duplication

[Kangji]

@taegeonum Thanks for the review! I've addressed your comments.

[taegeonum]

@Kangji I did another pass.

[taegeonum]

This explanation is not clear to me. Does this property set the destination executor for the output of intermediate vertex?

[wonook]

It limits the sources of the data that each task reads from, depending on where the task is located at. I'll add the explanation.

[taegeonum]

As far as I know, GroupByKey Transform is not always be represented as Combine PerKey Transform, so changing the name is confusing to me. For instance, CoGroupByKey is not combining, but it is also represented as GroupByKey as far as I know.

[taegeonum]

@Kangji Any update?

[Kangji]

@Kangji Any update?

not yet... :( It has been delayed due to the fall semester, even though i'm trying to do asap. I'll let you know.

[sonarqubecloud]

Kudos, SonarCloud Quality Gate passed!   

0 Bugs
0 Vulnerabilities
0 Security Hotspots
0 Code Smells

No Coverage information
0.0% Duplication

[wonook]

@taegeonum Can you take a final look?