Flink各种报错汇总及解决方法

这篇文章主要用来记录一下Flink中的常见的报错以及解决方案(以后会持续更新)

1,Table is not an append-only table. Use the toRetractStream() in order to handle add and retract messages.

这个是因为动态表不是append-only模式的,需要用toRetractStream(回撤流)处理就好了.

tableEnv.toRetractStream[Person](result).print()

2.今天在启动Flink任务的时候报错Caused by: java.lang.RuntimeException: Couldn't deploy Yarn cluster,然后仔细看发现里面有这么一句话system times on machines may be out of sync,意思说是机器上的系统时间可能不同步.

(1)安装ntpdate工具

yum -y install ntp ntpdate

(2)设置系统时间与网络时间同步

ntpdate cn.pool.ntp.org

在三台机器上分别执行完这个,在启动任务,发现可以了.

3.Could not retrieve the redirect address of the current leader. Please try to refresh,Flink任务在运行了一段时间后,进程还在但是刷新UI界面提示报错,你把这个job杀掉,重新启动,还是会报这个错.

解决方法是把这个目录下的文件删除,重启就可以了

high-availability.zookeeper.path.root: /flink

ZooKeeper节点根目录，其下放置所有集群节点的namespace。

4.No data sinks have been created yet. A program needs at least one sink that consumes data. Examples are writing the data set or printing it.

这个错是因为没有sink,解决方法是execute执行前面加上Sink就好了,例如:writeAsText

5.could not find implicit value for evidence parameter of type org.apache.flink.api.common.typeinfo.TypeInformation[String]

这个错是因为scala的隐式转换的问题. 导入这个包就好了.import org.apache.flink.streaming.api.scala._

6,有时候在重启Flink作业的时候会抛出异常,Wrong / missing exception when submitting job,这个实际是Flink的一个BUG,已经修复了,具体的情况可以看下jira, https://issues./jira/browse/FLINK-10312

7,Flink启动的时候有时候会报这个错,

2019-02-23 07:27:53,093 ERROR org.apache.flink.runtime.rest.handler.job.JobSubmitHandler    - Implementation error: Unhandled exception.
akka.pattern.AskTimeoutException: Ask timed out on [Actor[akka://flink/user/dispatcher#1998075247]] after [10000 ms]. Sender[null] sent message of type "org.apache.flink.runtime.rpc.messages.LocalFencedMessage".
	at akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(AskSupport.scala:604)
	at akka.actor.Scheduler$$anon$4.run(Scheduler.scala:126)
	at scala.concurrent.Future$InternalCallbackExecutor$.unbatchedExecute(Future.scala:601)
	at scala.concurrent.BatchingExecutor$class.execute(BatchingExecutor.scala:109)
at scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:599)
at akka.actor.LightArrayRevolverScheduler$TaskHolder.executeTask(LightArrayRevolverScheduler.scala:329)
at akka.actor.LightArrayRevolverScheduler$$anon$4.executeBucket$1(LightArrayRevolverScheduler.scala:280)
at akka.actor.LightArrayRevolverScheduler$$anon$4.nextTick(LightArrayRevolverScheduler.scala:284)
at akka.actor.LightArrayRevolverScheduler$$anon$4.run(LightArrayRevolverScheduler.scala:236)
at java.lang.Thread.run(Thread.java:745)
2019-02-23 07:27:54,156 ERROR org.apache.flink.runtime.rest.handler.legacy.files.StaticFileServerHandler  - Could not retrieve the redirect address.
java.util.concurrent.CompletionException: akka.pattern.AskTimeoutException: Ask timed out on [Actor[akka://flink/user/dispatcher#1998075247]] after [10000 ms]. Sender[null] sent message of type "org.apache.flink.runtime.rpc.messages.LocalFencedMessage".
at java.util.concurrent.CompletableFuture.internalComplete(CompletableFuture.java:205)
at java.util.concurrent.CompletableFuture$ThenApply.run(CompletableFuture.java:723)
at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:193)
at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2361)
at org.apache.flink.runtime.concurrent.FutureUtils$1.onComplete(FutureUtils.java:772)
at akka.dispatch.OnComplete.internal(Future.scala:258)
at akka.dispatch.OnComplete.internal(Future.scala:256)
at akka.dispatch.japi$CallbackBridge.apply(Future.scala:186)
at akka.dispatch.japi$CallbackBridge.apply(Future.scala:183)
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36)
at org.apache.flink.runtime.concurrent.Executors$DirectExecutionContext.execute(Executors.java:83)
at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:44)
at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:252)
at akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(AskSupport.scala:603)
at akka.actor.Scheduler$$anon$4.run(Scheduler.scala:126)
at scala.concurrent.Future$InternalCallbackExecutor$.unbatchedExecute(Future.scala:601)
at scala.concurrent.BatchingExecutor$class.execute(BatchingExecutor.scala:109)
at scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:599)
at akka.actor.LightArrayRevolverScheduler$TaskHolder.executeTask(LightArrayRevolverScheduler.scala:329)
at akka.actor.LightArrayRevolverScheduler$$anon$4.executeBucket$1(LightArrayRevolverScheduler.scala:280)
at akka.actor.LightArrayRevolverScheduler$$anon$4.nextTick(LightArrayRevolverScheduler.scala:284)
at akka.actor.LightArrayRevolverScheduler$$anon$4.run(LightArrayRevolverScheduler.scala:236)
at java.lang.Thread.run(Thread.java:745)

 找到了https://issues./jira/browse/FLINK-8485,说是Flink的bug,显示已经修复了,但是我感觉还是有点问题,还是会出现这个问题.

8.今天运行Flink的时候报了一个这个警告,找了半天没找到原因,最后发现了一个issues,其实这个是Flink的一个bug,在1.6.1版本已经修复了

org.apache.flink.runtime.filecache.FileCache  - improper use of releaseJob() without a matching number of createTmpFiles() calls for jobId b30ddc7e088eaf714e96a0630815440f

具体的情况看下面的连接: https://issues./jira/browse/FLINK-10283

9,Caused by: java.lang.RuntimeException: Rowtime timestamp is null. Please make sure that a proper TimestampAssigner is defined and the stream environment uses the EventTime time characteristic.

这句话的意思是说,Rowtime时间戳为null。请确保定义了正确的TimestampAssigner，并且流环境使用EventTime时间特性。就是说我们要先从数据源中提取时间戳,然后才能使用rowtime,在上面分配一下watermark就好了.

10,flink.table.TableJob$person$3(name: String, age: Integer, timestamp: Long)' must be static and globally accessible

这个报错是因为我们定义的case class类型必须是静态的,全局可访问的,就是说要把它放到main方法的外面就可以了

11,The proctime attribute can only be appended to the table schema and not replace an existing field. Please move 'proctime' to the end of the schema

这个报错的翻译为, proctime属性只能附加到表模式，而不能替换现有字段。请将'proctime'移到架构的末尾,我们在用proctime的时候要把他放到字段的最后一个位置.而不能放到其他的位置.