flink报错：No new data sinks have been defined since the last execution

男娘i 2021-09-25 02:34 362阅读 0赞

初学flink，在IDEA上编写WordCount代码运行报错如下：

Exception in thread "main" java.lang.RuntimeException: No new data sinks have been defined since the last execution. The last execution refers to the latest call to 'execute()', 'count()', 'collect()', or 'print()'.
    	at org.apache.flink.api.java.ExecutionEnvironment.createProgramPlan(ExecutionEnvironment.java:1092)
    	at org.apache.flink.api.java.ExecutionEnvironment.createProgramPlan(ExecutionEnvironment.java:1072)
    	at org.apache.flink.api.java.ExecutionEnvironment.executeAsync(ExecutionEnvironment.java:969)
    	at org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:891)
    	at org.apache.flink.api.scala.ExecutionEnvironment.execute(ExecutionEnvironment.scala:509)
    	at com.fuyun.flink.WordCount$.main(WordCount.scala:42)
    	at com.fuyun.flink.WordCount.main(WordCount.scala)

代码如下：

package com.fuyun.flink
    
    // 因为很多算子Java和Scala名称都一样，必须要显示导入scala所有包告知使用Scala语言的算子，不然会报错
    import org.apache.flink.api.scala._
    
    object BatchWordCount { 
    
      def main(args: Array[String]): Unit = { 
    
        // 定义输入输出文件路径
        val input = "E:\\IDEAworkspace\\flinkLearning\\datas\\WordCount"
    
        // 创建一个批处理的执行环境
        val env:ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
        // 读取离线文件
        val inputDataSet:DataSet[String] = env.readTextFile(input)
    
        // 对离线文件单词统计，不区分大小写
        val counts = inputDataSet.flatMap(_.toLowerCase.split("\\s")) // 对元素转换为小写再进行分割
          .filter(_.nonEmpty) // 过滤空值
          .map((_, 1))  // 将元素组合成二元组
          .groupBy(0)  // 对第一个元素进行分组
          .sum(1)  // 对第二个元素求和
    
         // 控制台答应结果
         counts.print()
         
         // 调用execute方法执行
         env.execute("bath word count")
    
      }
    }

问题原因及解决办法：  
从报错日志可看出，自上次执行以来，没有定义新的数据接收器。对于离线批处理的算子，如：“count()”、“collect()”或“print()”等既有sink功能，还有触发的功能。  
我们上面调用了print()方法，会自动触发execute，所以最后面的一行执行器没有数据可以执行。所以去掉最后一行代码即可，用了print()方法后无需再调用execute()方法。