提交flink程序到集群上

提交flink程序到集群上

这里我演示的是 yarn的cluster模式

首先来到IDEA中将咱们需要在环境上跑的程序，进行打包。
这里使用的IDEA上的图形界面进行项目打包的(还是挺方便的)

执行完上面三步后：项目的下面会出现target文件夹

这里产生两个jar包，如果测试环境上有项目所需的资源jar的话，可以直接将
较小的jar上传到环境上执行。

如果测试环境上没有程序所需资源jar包，则需要将较大的jar包上传到环境上执行。

两个jar之所以不一样大小，就是一个包含程序所需资源jar包，一个不包含程序所
需资源jar包。

将jar上传到一个目录下：

pwd 复制目录

集群上执行程序的命令：

我是在flink的bin目录下执行的程序
-m 执行执行模式
-d 后台运行
-c 全路径包名

./flink run 
-m yarn-cluster 
-d 
-c com.zw.bigdata.qy.logs.writer.PushLogToOssParquetWrite 
/opt/cloudera/parcels/Flink/lib/flink/examples/bigdata-cps-gzh-rt/cps-gzh-rt-1.0-SNAPSHOT-big.jar

如果资源不够用：

增加运行资源，重新执行。(如果不行可以再扩大至4096)
./flink run 
-m yarn-cluster -ys 1 -yjm 2048 -ytm 2048 
-d 
-c com.zw.bigdata.qy.logs.writer.PushLogToOssParquetWrite 
/opt/cloudera/parcels/Flink/lib/flink/examples/bigdata-cps-gzh-rt/2022-01-05/cps-gzh-rt-1.0-SNAPSHOT-big.jar

命令执行成功后：

记得在输出的日志中找到一个web的url
例如:node-01:44387
这个是用来访问flink的web客户端的

忘记了程序执行产生的访问端口：

tail -f flink-root-client-test001.log  
到flink的安装目录下进行查看本地日志，中会有相关信息
test003:44987 of application 'application_1639462740764_20402'  JobID ccc8c6b0b83ffc13b27bde276b04d4c4

如果flink客户端没有实际有用的日志：可到yarn上看日志

在web界面提交flink程序时报Failed to rollback to checkpoint/savepoint 怎么解

程序运行48小时会报一个异常：

--只是推测--
此异常可能是，数据中断导致。dwd层未消费到ods层的数据
(也就是说ods有一段时间没有产生数据[可能是数据源断了])

failed. Quarantining address.
java.util.concurrent.TimeoutException: Remote system has been silent for too long. (more than 48.0 hours)
at akka.remote.ReliableDeliverySupervisor$$anonfun$idle.applyOrElse(Endpoint.scala:387) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
at akka.actor.Actor.aroundReceive(Actor.scala:517) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
at akka.actor.Actor.aroundReceive$(Actor.scala:515) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
at akka.remote.ReliableDeliverySupervisor.aroundReceive(Endpoint.scala:207) ~[flink-dist_2.12-1.13.3.jar:1.13.3]
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592) [flink-dist_2.12-1.13.3.jar:1.13.3]
at akka.actor.ActorCell.invoke(ActorCell.scala:561) [flink-dist_2.12-1.13.3.jar:1.13.3]
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258) [flink-dist_2.12-1.13.3.jar:1.13.3]
at akka.dispatch.Mailbox.run(Mailbox.scala:225) [flink-dist_2.12-1.13.3.jar:1.13.3]
at akka.dispatch.Mailbox.exec(Mailbox.scala:235) [flink-dist_2.12-1.13.3.jar:1.13.3]
at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [flink-dist_2.12-1.13.3.jar:1.13.3]
at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [flink-dist_2.12-1.13.3.jar:1.13.3]
at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [flink-dist_2.12-1.13.3.jar:1.13.3]
at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [flink-dist_2.12-1.13.3.jar:1.13.3]