zoukankan      html  css  js  c++  java
  • Flink用Lambda表达式报错

    Flink用Lambda表达式报错

    Caused by: org.apache.flink.api.common.functions.InvalidTypesException: The generic type parameters of 'Collector' are missing. In many cases lambda methods don't provide enough information for automatic type extraction when Java generics are involved. An easy workaround is to use an (anonymous) class instead that implements the 'org.apache.flink.api.common.functions.FlatMapFunction' interface. Otherwise the type has to be specified explicitly using type information.

    大致意思是,lambda写法无法提供足够的类型信息,无法推断出正确的类型,建议要么改成匿名类写法,要么用type information提供明细的类型信息。

    而且官网是也是有说明的,Java Lambda Expressions | Apache Flink,于是参考了仿写了一下。

    /**
     * @author WGR
     * @create 2021/7/31 -- 22:00
     */
    public class WordCount {
    
        public static void main(String[] args)  throws Exception{
            //1.准备环境
            ExecutionEnvironment environment = ExecutionEnvironment.getExecutionEnvironment();
    
            //2.准备数据
            DataSet<String> dataSet = environment.fromElements("dalianpai hello flink", "dalianpai hello kafka", "dalianpai");
    
            //3.处理数据
            DataSet<String> words = dataSet.flatMap(new FlatMapFunction<String, String>() {
                @Override
                public void flatMap(String value, Collector<String> out) throws Exception {
                    //value表示每一行数据
                    String[] arr = value.split(" ");
                    for (String word : arr) {
                        out.collect(word);
                    }
                }
            });
    
            DataSet<Tuple2<String, Integer>> wordAndOne = words.map(new MapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public Tuple2<String, Integer> map(String value) throws Exception {
                    //value就是每一个单词
                    return Tuple2.of(value, 1);
                }
            });
    
            //分组
            UnsortedGrouping<Tuple2<String, Integer>> grouped = wordAndOne.groupBy(0);
    
            //聚合
            AggregateOperator<Tuple2<String, Integer>> result = grouped.sum(1);
    
            //TODO 3.sink
            result.print();
        }
    }
    
    

    lambda表达式:

    /**
     * @author WGR
     * @create 2021/7/31 -- 22:23
     */
    public class WordCount2 {
    
        public static void main(String[] args)  throws Exception {
            ExecutionEnvironment.getExecutionEnvironment()
                    .fromElements("dalianpai hello flink", "dalianpai hello kafka", "dalianpai")
                    .flatMap((FlatMapFunction<String, String>)(value, out) -> {
                        //value表示每一行数据
                        String[] arr = value.split(" ");
                        for (String word : arr) {
                            out.collect(word);
                        }
                    }).returns(Types.STRING)
                    .map( i -> Tuple2.of(i, 1)).returns(Types.TUPLE(Types.STRING, Types.INT))
                    .groupBy(0)
                    .sum(1)
                    .print();
        }
    }
    
  • 相关阅读:
    [WWF]相关资料整理
    [CSharp]经常遇到字符编码转换的问题
    [BizTalk]BizTalk Server 2004的监控和管理
    [Linux]删除目录
    [BizTalk]好用的BizTalk管理工具
    [JWF]使用脚本访问ActiveDirectory(一)ADODB
    [BizTalk][Why]持久化数据库的状态时出现异常
    [BizTalk]关于BizTalk中的命名规则
    FTP连接
    [导入]跟踪BTSWebSvcWiz.exe的执行过程
  • 原文地址:https://www.cnblogs.com/dalianpai/p/15085599.html
Copyright © 2011-2022 走看看