Flink提交任务的三种方式 Flink提交任务的三种方式web页面提交 flink命令提交将jar包上传至集群，提交任务flink run -c com.flink.Flinktest flinktest.jar com.flink.Flinktest -- 主类名flinktest.jar -- jar包名rpc方式提交任务 --- 远程提交用的较少代码写完之后需要先打包，再运行Flink框架报错有一个特点：前几条报错原因都是废话，要从后面看package com.flink import org.apache.flink.streaming.api.scala._ object Flinktest { def main(args: Array[String]): Unit = { /** * 创建远程环境，远程提交flink任务 * */ val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createRemoteEnvironment( //主机名 "master", //端口号 8081, //指定jar包的路径 "C:\\Users\\HB\\Desktop\\market_analysis\\flinktest.jar" ) val linesDS: DataStream[String] = env.socketTextStream("master", 8888) //1、将数据展开 val wordsDS: DataStream[String] = linesDS.flatMap(line => line.split(",")) //2、转换成kv格式 val kvDS: DataStream[(String, Int)] = wordsDS.map((_, 1)) //3、按照单词进行分组 val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(kv => kv._1) //4、统计数量,对value进行求和, 指定下标进行聚合 val countDS: DataStream[(String, Int)] = keyByDS.sum(1) //打印结果 countDS.print() env.execute("rpc") } }

Flink提交参数详解 Flink命令行提交参数1 参数必选 ： -n,--container <arg> 分配多少个yarn容器 (=taskmanager的数量) 2 参数可选 ： -D <arg> 动态属性 -d,--detached 独立运行 -jm,--jobManagerMemory <arg> JobManager的内存 [in MB] -nm,--name 在YARN上为一个自定义的应用设置一个名字 -q,--query 显示yarn中可用的资源 (内存, cpu核数) -qu,--queue <arg> 指定YARN队列. -s,--slots <arg> 每个TaskManager使用的slots数量 -tm,--taskManagerMemory <arg> 每个TaskManager的内存 [in MB] -z,--zookeeperNamespace <arg> 针对HA模式在zookeeper上创建NameSpace -id,--applicationId <yarnAppId> YARN集群上的任务id，附着到一个后台运行的yarn session中 3 run [OPTIONS] <jar-file> <arguments> run操作参数: -c,--class <classname> 如果没有在jar包中指定入口类，则需要在这里通过这个参数指定 -m,--jobmanager <host:port> 指定需要连接的jobmanager(主节点)地址，使用这个参数可以指定一个不同于配置文件中的jobmanager -p,--parallelism <parallelism> 指定程序的并行度。可以覆盖配置文件中的默认值。 4 启动一个新的yarn-session,它们都有一个y或者yarn的前缀 例如：./bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar 连接指定host和port的jobmanager： ./bin/flink run -m SparkMaster:1234 ./examples/batch/WordCount.jar -input hdfs://hostname:port/hello.txt -output hdfs://hostname:port/result1 启动一个新的yarn-session： ./bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar -input hdfs://hostname:port/hello.txt -output hdfs://hostname:port/result1 5 注意：命令行的选项也可以使用./bin/flink 工具获得。 6 Action "run" compiles and runs a program. Syntax: run [OPTIONS] <jar-file> <arguments> "run" action options: -c,--class <classname> Class with the program entry point ("main" method or "getPlan()" method. Only needed if the JAR file does not specify the class in its manifest. -C,--classpath <url> Adds a URL to each user code classloader on all nodes in the cluster. The paths must specify a protocol (e.g. file://) and be accessible on all nodes (e.g. by means of a NFS share). You can use this option multiple times for specifying more than one URL. The protocol must be supported by the {@link java.net.URLClassLoader}. -d,--detached If present, runs the job in detached mode -n,--allowNonRestoredState Allow to skip savepoint state that cannot be restored. You need to allow this if you removed an operator from your program that was part of the program when the savepoint was triggered. -p,--parallelism <parallelism> The parallelism with which to run the program. Optional flag to override the default value specified in the configuration. -q,--sysoutLogging If present, suppress logging output to standard out. -s,--fromSavepoint <savepointPath> Path to a savepoint to restore the job from (for example hdfs:///flink/savepoint-1537). 7 Options for yarn-cluster mode: -d,--detached If present, runs the job in detached mode -m,--jobmanager <arg> Address of the JobManager (master) to which to connect. Use this flag to connect to a different JobManager than the one specified in the configuration. -yD <property=value> use value for given property -yd,--yarndetached If present, runs the job in detached mode (deprecated; use non-YARN specific option instead) -yh,--yarnhelp Help for the Yarn session CLI. -yid,--yarnapplicationId <arg> Attach to running YARN session -yj,--yarnjar <arg> Path to Flink jar file -yjm,--yarnjobManagerMemory <arg> Memory for JobManager Container with optional unit (default: MB) -yn,--yarncontainer <arg> Number of YARN container to allocate (=Number of Task Managers) -ynl,--yarnnodeLabel <arg> Specify YARN node label for the YARN application -ynm,--yarnname <arg> Set a custom name for the application on YARN -yq,--yarnquery Display available YARN resources (memory, cores) -yqu,--yarnqueue <arg> Specify YARN queue. -ys,--yarnslots <arg> Number of slots per TaskManager -yst,--yarnstreaming Start Flink in streaming mode -yt,--yarnship <arg> Ship files in the specified directory (t for transfer) -ytm,--yarntaskManagerMemory <arg> Memory per TaskManager Container with optional unit (default: MB) -yz,--yarnzookeeperNamespace <arg> Namespace to create the Zookeeper sub-paths for high availability mode -z,--zookeeperNamespace <arg> Namespace to create the Zookeeper sub-paths for high availability modeflink run参数flink run命令执行模板：flink run [option]-c,–class : 需要指定的main方法的类-C,–classpath : 向每个用户代码添加url，他是通过UrlClassLoader加载。url需要指定文件的schema如（file://）-d,–detached : 在后台运行-p,–parallelism : job需要指定env的并行度，这个一般都需要设置。-q,–sysoutLogging : 禁止logging输出作为标准输出。-s,–fromSavepoint : 基于savepoint保存下来的路径，进行恢复。-sae,–shutdownOnAttachedExit : 如果是前台的方式提交，当客户端中断，集群执行的job任务也会shutdown。flink run -m yarn-cluster参数-m,–jobmanager : yarn-cluster集群-yd,–yarndetached : 后台-yjm,–yarnjobManager : jobmanager的内存-ytm,–yarntaskManager : taskmanager的内存-yn,–yarncontainer : TaskManager的个数-yid,–yarnapplicationId : job依附的applicationId-ynm,–yarnname : application的名称-ys,–yarnslots : 分配的slots个数例：flink run -m yarn-cluster -yd -yjm 1024m -ytm 1024m -ynm -ys 1flink-listflink list：列出flink的job列表。flink list -r/–runing :列出正在运行的jobflink list -s/–scheduled :列出已调度完成的jobflink cancelflink cancel [options] <job_id> : 取消正在运行的job idflink cancel -s/–withSavepoint <job_id> ： 取消正在运行的job，并保存到相应的保存点通过 -m 来指定要停止的 JobManager 的主机地址和端口例： bin/flink cancel -m 127.0.0.1:8081 5e20cb6b0f357591171dfcca2eea09deflink stop :仅仅针对Streaming jobflink stop [options] <job_id>flink stop <job_id>：停止对应的job通过 -m 来指定要停止的 JobManager 的主机地址和端口例： bin/flink stop -m 127.0.0.1:8081 d67420e52bd051fae2fddbaa79e046bb取消和停止（流作业）的区别如下：　　cancel() 调用，立即调用作业算子的 cancel() 方法，以尽快取消它们。如果算子在接到 cancel() 调用后没有停止，Flink 将开始定期中断算子线程的执行，直到所有算子停止为止。　　stop() 调用，是更优雅的停止正在运行流作业的方式。stop() 仅适用于 Source 实现了 StoppableFunction 接口的作业。当用户请求停止作业时，作业的所有 Source 都将接收 stop() 方法调用。直到所有 Source 正常关闭时，作业才会正常结束。这种方式，使作业正常处理完所有作业。flink modify 修改任务并行度flink modify <job_id> [options]flink modify <job_id> -p /–parallelism p : 修改job的并行度例： flink modify -p 并行数 <job_pid>flink savepointflink savepoint [options] <job_id>eg: # 触发保存点flink savepoint <job_id> hdfs://xxxx/xx/x : 将flink的快照保存到hdfs目录使用yarn触发保存点flink savepoint <job_id> <target_directory> -yid <application_id>使用savepoint取消作业flink cancel -s <tar_directory> <job_id>从保存点恢复flink run -s <target_directoey> [:runArgs]如果复原的程序，对逻辑做了修改，比如删除了算子可以指定allowNonRestoredState参数复原。flink run -s <target_directory> -n/–allowNonRestoredState [:runArgs]savepoint 与 checkpoint 的区别　　checkpoint是增量做的，每次的时间短，数据量小，只要在程序里面启用后会自动触发，用户无需感知；savepoint是全量做的，时间长，数据量大，需要用户主动触发。　　checkpoint 是作业failover 的时候自动使用，不需要用户指定，savepoint 一般用于程序版本更新、bug修复、A/B Test 等场景，需要用户指定。

Flink集群搭建--Yarn模式 准备Flink ON YARN模式就是使用客户端的方式，直接向Hadoop集群提交任务即可。不需要单独启动Flink进程。注意：1。Flink ON YARN 模式依赖Hadoop 2.4.1及以上版本2.Flink ON YARN支持两种使用方式 yarn-session会话模式:　使用Flink中的yarn-session（yarn客户端），会启动两个必要服务 JobManager 和 TaskManagers　客户端通过yarn-session提交作业　yarn-session会一直启动，不停地接收客户端提交的作用　有大量的小作业，适合使用这种方式yarn-per-job模式　直接提交任务给YARN，大作业，适合使用这种方式安装包下载：http://archive.apache.org/dist/flink/flink-1.10.2/ 　Flink on Yarn集群搭建{dotted startColor="#ff6c6c" endColor="#1989fa"/}yarn-session模式搭建解压[root@master ~]# tar -zxvf /data/flink-1.10.2 -C /usr/local/src [root@master ~]# mv /usr/local/src/flink-1.10.2 /usr/local/src/flink配置环境变量[root@master ~]# vi /etc/profile export FLINK_HOME=/usr/local/src/flink export PATH=$PATH:$FLINK_HOME/bin [root@master ~]# source /etc/profileflink on yarn搭建，资源的调度由yarn完成，所以不需要修改配置文件HADOOP_CLASSPATH　　执行yarn-session.sh脚本需要先设置HADOOP_CLASSPATH这个环境变量，否则，执行yarn-session.sh报错，提示找不到hadoop的一些依赖方式一(推荐):在/etc/profile中配置HADOOP_CLASSPATH[root@master ~]# vi /etc/profile export HADOOP_CLASSPATH=`${HADOOP_HOME}/bin/hadoop classpath`方式二:命令行导入HADOOP_CLASSPATH[root@master ~]# export HADOOP_CLASSPATH=`hadoop classpath`缺点：只在当前命令行本次运行有效方式三:修改yarn-session.sh，添加HADOOP_CLASSPATH启动集群接下来，使用yarn-session.sh在YARN中创建一个长时间运行的Flink集群[root@master flink]# bin/yarn-session.sh -jm 1024m -tm 1024m -d这个表示创建一个Flink集群，-jm是指定主节点的内存，-tm是指定从节点的内存，-d是表示把这个进程放到后台去执行。启动之后，会看到类似这样的日志信息，这里面会显示flink web界面的地址，以及这个flink集群在yarn中对应的applicationid 此时在YARN的web界面中可以看到这个flink集群。 可以使用屏幕中显示的flink的web地址或者yarn中这个链接都是可以进入这个flink的web界面 提交作业(job)接下来向这个Flink集群中提交任务，此时使用Flink中的内置案例。[root@master flink]# bin/flink run ./examples/batch/WordCount.jar注意：这个时候我们使用flink run的时候，它会默认找这个文件，然后根据这个文件找到刚才我们创建的那个永久的Flink集群，这个文件里面保存的就是刚才启动的那个Flink集群在YARN中对应的applicationid。[root@master flink]# more /tmp/.yarn-properties-root #Generated YARN properties file #Tue Jun 07 01:39:29 CST 2022 dynamicPropertiesString= applicationID=application_1654423542500_0004任务提交上去执行完成之后，再来看flink的web界面，发现这里面有一个已经执行结束的任务了。 注意：这个任务在执行的时候，会动态申请一些资源执行任务，任务执行完毕之后，对应的资源会自动释放掉。最后把这个Flink集群停掉，使用yarn的kill命令。[root@master flink]# yarn application -kill application_1654423542500_0004 Killing application application_1654423542500_0004 22/06/07 01:46:43 INFO impl.YarnClientImpl: Killed application application_1654423542500_0004针对yarn-session命令，它后面还支持一些其它参数，可以在后面传一个-help参数[root@master flink]# bin/yarn-session.sh -help Usage: Optional -at,--applicationType <arg> Set a custom application type for the application on YARN -D <property=value> use value for given property -d,--detached If present, runs the job in detached mode -h,--help Help for the Yarn session CLI. -id,--applicationId <arg> Attach to running YARN session -j,--jar <arg> Path to Flink jar file -jm,--jobManagerMemory <arg> Memory for JobManager Container with optional unit (default: MB) -m,--jobmanager <arg> Address of the JobManager to which to connect. Use this flag to connect to a different JobManager than the one specified in the configuration. -nl,--nodeLabel <arg> Specify YARN node label for the YARN application -nm,--name <arg> Set a custom name for the application on YARN -q,--query Display available YARN resources (memory, cores) -qu,--queue <arg> Specify YARN queue. -s,--slots <arg> Number of slots per TaskManager -t,--ship <arg> Ship files in the specified directory (t for transfer) -tm,--taskManagerMemory <arg> Memory per TaskManager Container with optional unit (default: MB) -yd,--yarndetached If present, runs the job in detached mode (deprecated; use non-YARN specific option instead) -z,--zookeeperNamespace <arg> Namespace to create the Zookeeper sub-paths for high availability mode常见命令的中文注释 yarn-per-job模式搭建flink run -m yarn-cluster (创建Flink集群+提交任务)使用flink run直接创建一个临时的Flink集群，并且提交任务此时这里面的参数前面加上了一个y参数[root@master flink]# bin/flink run -m yarn-cluster -yjm 1024 -ytm 1024 ./examples/batch/WordCount.jar针对Flink命令的一些用法汇总

Flink集群搭建--Local模式 前期准备安装包下载：http://archive.apache.org/dist/flink/flink-1.10.2/ 本地模式安装{dotted startColor="#ff6c6c" endColor="#1989fa"/}解压[root@master ~]# tar -zxvf /data/flink-1.10.2 -C /usr/local/src/ [root@master ~]# mv /usr/local/src/flink-1.10.2 /usr/local/src/flink配置环境变量[root@master ~]# vi /etc/profile export FLINK_HOME=/usr/local/src/flink export PATH=$PATH:$FLINK_HOME/bin [root@master ~]# source /etc/profile启动集群[root@master]# start-cluster.sh提交作业(job)[root@master ~]# flink run $FLINK_HOME/examples/streaming/WordCount.jar查看日志(结果)[root@master ~]# tail $FLINK_HOME/log/flink-*-taskexecutor-*.out监视集群的状态和正在运行的作业http://localhost:8081/#/overview 停止集群[root@master ~]# stop-cluster.sh