Flink中指定算子的方式是什么

今天就跟大家聊聊有關(guān)Flink中指定算子的方式是什么，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

我們之前在使用flatMap時(shí)，傳了一個(gè)new FlatMapFunction匿名內(nèi)部類。而這僅僅是其中的一種方式。

方式一：實(shí)現(xiàn)MapFunction接口

最簡(jiǎn)單的方式就是實(shí)現(xiàn)一個(gè)MapFunction接口，例如：

text.flatMap(new MyFlatMapFunction()).keyBy(new KeySelector<WC, Object>() {
            @Override
            public Object getKey(WC value) throws Exception {
                return value.word;
            }
        }).timeWindow(Time.seconds(5)).sum("count").print().setParallelism(1);
public static class MyFlatMapFunction implements FlatMapFunction<String, WC> {
        @Override
        public void flatMap(String value, Collector<WC> out) throws Exception {
            String[] tokens = value.toLowerCase().split(",");
            for (String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new WC(token, 1));
                }
            }
        }
    }

方式二：匿名內(nèi)部類

這種方式就是我們之前一直使用的方式。

方式三：java8 Lambdas

方式四：Rich functions

text.flatMap(new RichFlatMapFunction<String, WC>() {
            @Override
            public void flatMap(String value, Collector<WC> out) throws Exception {
                String[] tokens = value.toLowerCase().split(",");
                for (String token : tokens) {
                    if (token.length() > 0) {
                        out.collect(new WC(token, 1));
                    }
                }
            }
        })

繼承一個(gè)RichFlatMapFunction類

看完上述內(nèi)容，你們對(duì)Flink中指定算子的方式是什么有進(jìn)一步的了解嗎？如果還想了解更多知識(shí)或者相關(guān)內(nèi)容，請(qǐng)關(guān)注億速云行業(yè)資訊頻道，感謝大家的支持。