[FLINK-1622][java-api][scala-api] add a GroupCombine operator

[mxm]

The GroupCombine operator acts like a the optional combine step in the
GroupReduceFunction. It is more general because it combines from an
input to an arbitrary output type. Combining is performed on the
partitions with as much data in memory as possible. This may lead to
partial results.

The operator can be used to pre-combine elements into an intermediate
output format before applying a proper groupReduce to produce the final
output format.

[hsaputra]

@mxm, could you add class description on why the base class need to be created and how it relates to existing ones?
We would love to add more documentation on the code to help new contributors get comfortable with the code flow.

[mxm]

Thanks for spotting the missing doc in this class. I'll add some.

[aljoscha]

I like the implementation, except for my comments on groupReducePartial() on grouped DataSets. Also, the tests seem a bit shady because of all the grouping and regular reduceGroup operations. I would suggest partitioning the data using a manual partition operation and then applying a GroupReducePartial.

[hsaputra]

I assume this will be new operator but I do not see updates on the documentation files.

[mxm]

@aljoscha Thanks for the comments. I agree, the tests are a bit shady because they test the operator by first performing a partial, then a full reduce. Using a custom partitioner would make more sense.

@hsaputra Yes, we absolutely have to add docs apart from the Java/Scala docs.

What do you think about the name? @fhueske suggested to expose the operator as combine because it is essentially a user-accessible combiner..

[fhueske]

Yes, I've got a couple of comments as well.

First of all, as @mxm said, I would propose to call this operator combine because it is a generalized combiner (output type may be different from input type). The current combiner could be implemented as special case of this one (input = output type).

I think @aljoscha is right. partialGroupReduce (or combine) should be applied on a DataSet, not a GroupedDataSet. Otherwise, you lose most of the benefits (local processing) and would ship all data which is exactly what you want to avoid when using a combiner. If you want to run a reduceafterwards, you simply do groupBy(x).reduceGroup().

Apart from documentation, also a few Scala tests should be added.

[fhueske]

Isnt't the partial GroupReduce only working on subsets of a local partition (as much as fits in the sort buffer)?

[mxm]

That is correct.

[fhueske]

I have to correct myself. A combiner should of course be called on groups of records. Therefore, calling it on a Grouping makes absolute sense. However, the semantics of the Grouping are different for Combine and Reduce. For Reduce a full Grouping is established with repartitioning and full sort, whereas for Combine only a partial local grouping should be created.

[mxm]

@aljoscha @fhueske For a general combine, the operator can be used without grouping. When we want to combine elements before performing a proper groupReduce with a groupBy, we also need to use groupBy for the combine. Otherwise, we wouldn't know in the combiner, which keys belong together. However, there are cases where a combiner without a groupBy would be appriopriate. That's why the new operator is exposed in DataSet, GroupedDataSet, ad SortedDataSet. We currently have the same behavior for a normal GroupReduce despite the fact that the combine of the GroupReduce cannot change the input type.

[aljoscha]

Sorry, I completely blanked, of course, You still need the grouping, only the shuffle step you don't need.

So, I suggest only better tests, using a combination of partitionByHash() and groupReducePartial().

[mxm]

@aljoscha @fhueske @hsaputra Thanks for the feedback. Some people suggested that the name is confusing and that my pull request involved too much code duplication. I propose to call the new operator combineGroup because it is a combiner that works on Groups, just like the existing combiners in GroupReduceCombineDriver and AllGroupReduceDriver. I refactored the code to reuse Flink's existing logic for combiners.

@hsaputra I added documentation for the operator in the code and the official documentation. Some Scala tests for the API were added as well.

When merging this pull request, I would squash the two commits and keep the commit message of the latest one. I simply wanted to keep the first one to show the development process.

[hsaputra]

Could you add JavaDoc for this class? I know other derived classes for OperatorDescriptorSingle does not have JavaDoc, yet but I am trying to get new code added to have at least simple explanation why the class should be created.

[mxm]

Totally agree with you. We should not accept undocumented classes. I added some doc, also for the base class.

[hsaputra]

Thanks @mxm ! I add couple more comments about documentation. I am trying to promote habit of adding JavaDoc comment for new classes to help indicate why it has to be created. Reduction of code is harder to do than adding new ones.

[mxm]

@hsaputra Thanks for the feedback. Promoting the habit of documenting every class is a very good thing.

[hsaputra]

HI @mxm, appreciate the ACK. Just trying to keep up all informed with more code coming in.
It is harder to read other people code and flow so was just trying to ease up the process =)

[mxm]

Any further comments?

[aljoscha]

I would say it's good to go now.