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

初学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()方法。

flink报错：No new data sinks have been defined since the last execution相关推荐

【Flink】No new data sinks have been defined since the last execution
在编写Flink(1.9)批处理任务Demo的时候遇到报错,报错详情: Exception in thread "main" java.lang.RuntimeException: ...
Flink学习笔记（一）：No new data sinks have been defined since the last execution.
问题描述: 初学flink,在IDEA中写离线批处理的wordcount后,运行报了以下异常: Exception in thread "main" java.lang.Runti ...
No new data sinks have been defined since the last execution.
完整报错如下: Exception in thread "main" java.lang.RuntimeException: No new data sinks have been ...
No new data sinks have been defined since the last execution
报错: Exception in thread "main" java.lang.RuntimeException: No new data sinks have been def ...
Flink 报错：unable to generate a JAAS configuration file
安装 Flink 报错:unable to generate a JAAS configuration file 原因是Flink的存放临时文件的地方,默认是 tmp/,例如我的是/data/flin ...
【Flink】Flink 报错 Writing records to streamload failed
1.概述 flink报错这个问题是 FLink是高频写入doris导致的
flink报错：The types of the interface org.apache.flink.util.OutputTag could not be inferred.
测试flink报错如下: The types of the interface org.apache.flink.util.OutputTag could not be inferred. Suppo ...
mysql报错：1406, Data too long for column
mysql报错:1406, "Data too long for column pymysql.err.DataError: (1406, "Data too long for c ...
.net调用Oracle 报错：System.Data.OracleClient 需要 Oracle 客户端软件 8.1.7 或更高版本问题
我又重拾.net搞一搞.老本行了,不过现在已投奔java阵营. 用visual studio 2022运行asp.net程序,结果在读取数据库的时候报错: .net调用Oracle 报错:System ...

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

flink报错：No new data sinks have been defined since the last execution相关推荐

最新文章

热门文章