[FLINK-6203] DataSet Transformations

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 1.2.0, 1.3.0
• Fix Version/s: 1.2.1, 1.3.0
• Component/s: API / DataSet, Documentation
• Assignee: Chesnay Schepler
• Reporter: 苏拓
• Created: 28/Mar/17 09:30
• Updated: 05/Apr/17 21:42
• Resolved: 05/Apr/17 21:42

Description

the example of GroupReduce on sorted groups can't remove duplicate Strings in a DataSet.
need to add "prev=t"
such as:

val output = input.groupBy(0).sortGroup(1, Order.ASCENDING).reduceGroup {
      (in, out: Collector[(Int, String)]) =>
        var prev: (Int, String) = null
        for (t <- in) {
          if (prev == null || prev != t)
            out.collect(t)
            prev=t // this line is missing in the example
        }
    }