[FLINK-11706] Add the SumFunction to support KeyedStream.sum with field which is array

• Type: Improvement
• Status: Closed
• Priority: Minor
• Resolution: Won't Do
• Component/s: API / DataStream
• Labels: pull-request-available
• Assignee: wangpeibin
• Reporter: wangpeibin
• Created: 21/Feb/19 07:07
• Updated: 07/Mar/19 14:36
• Resolved: 07/Mar/19 14:36

Description

The goal is to implement a KeyedStream API to sum with the field which is array.

The example code with like:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Tuple2<Long, Integer[]>> src = env.fromCollection(Arrays.asList(
   new Tuple2<>(1L, new Integer[]{2, 4}),
   new Tuple2<>(1L, new Integer[]{3, 6}),
   new Tuple2<>(1L, new Integer[]{4, 8}),
   new Tuple2<>(2L, new Integer[]{2, 4}),
   new Tuple2<>(2L, new Integer[]{3, 6}),
   new Tuple2<>(2L, new Integer[]{4, 8})
));

src.keyBy(0)
.sum(1}) 
.print();
env.execute();

the expected output should be 

(1,[2, 4]) 
(1,[5, 10])
(1,[9, 18])
(2,[2, 4])
(2,[5, 10])
(2,[9, 18])

right now this job will throw out exception:

java.lang.RuntimeException: DataStream cannot be summed because the class Integer[] does not support the + operator.

Issue Links

• links to: GitHub Pull Request #7786