Flink On Yarn 架构
前提条件
首先需要配置YARN_CONF_DIR, HADOOP_CONF_DIR ,HADOOP_CONF_PATH其中一个用来确保Flink能够访问HDFS和Yarn的RM。
主要启动流程
1. 启动进程
首先我们通过下面的命令行启动flink on yarn的集群
bin/yarn-session.sh -n 3 -jm 1024 -nm 1024 -st
这里将产生总共五个进程
- ** 1个FlinkYarnSessionCli ---> Yarn Client **
- ** 1个YarnApplicationMasterRunner ---> AM + JobManager**
- 3个YarnTaskManager --> TaskManager
即一个客户端+4个container,1个container启动AM,3个container启动TaskManager。
2.启动流程
- FlinkYarnSessionCli 启动的过程中首先会检查Yarn上有没有足够的资源去启动所需要的container,如果有,则上传一些flink的jar和配置文件到HDFS,这里主要是启动AM进程和TaskManager进程的相关依赖jar包和配置文件。
- 2.接着yarn client会首先向RM申请一个container来 ApplicationMaster(YarnApplicationMasterRunner进程),然后RM会通知其中一个NM启动这个container,被分配到启动AM的NM会首先去HDFS上下载第一步上传的jar包和配置文件到本地,接着启动AM;在这个过程中会启动JobManager,因为JobManager和AM在同一进程里面,它会把JobManager的地址重新作为一个文件上传到HDFS上去,TaskManager在启动的过程中也会去下载这个文件获取JobManager的地址,然后与其进行通信;AM还负责Flink的web 服务,Flink里面用到的都是随机端口,这样就允许了用户能够启动多个yarn session。
启动命令:
- 3.AM 启动完成以后,就会向AM申请container去启动TaskManager,启动的过程中也是首先从HDFS上去下载一些包含TaskManager(yarn模式的话这里就是YarnTaskManager )主类 的jar和启动过程依赖的配置文件,如JobManager地址所在的文件,然后利用java cp的方式去启动YarnTaskManager ,一旦这些准备好,就可以接受任务了。这个和spark on yarn的yarn cluster模式其实差不多,也是分为两个部分,一个是准备工人和工具(spark是启动sc的过程,flink是初始化ENV的过程),另外一个就是给工人分配具体工作(都是执行具体的操作,action什么的触发)。
启动命令:
进程信息
- ** FlinkYarnSessionCli **
/home/hadoop/ym/jdk1.8.0_101/bin/java -Xmx512m -classpath /home/hadoop/ym/flink-1.1.3/lib/flink-dist_2.10-1.1.3.jar:/home/hadoop/ym/flink-1.1.3/lib/flink-python_2.10-1.1.3.jar:/home/hadoop/ym/flink-1.1.3/lib/log4j-1.2.17.jar:/home/hadoop/ym/flink-1.1.3/lib/slf4j-log4j12-1.7.7.jar::/home/hadoop/ym/hadoop-2.7.1/etc/hadoop: -Dlog.file=/home/hadoop/ym/flink-1.1.3/log/flink-xxxuser-yarn-session-db-180.photo.163.org.log -Dlog4j.configuration=file:/home/hadoop/ym/flink-1.1.3/conf/log4j-yarn-session.properties -Dlogback.configurationFile=file:/home/hadoop/ym/flink-1.1.3/conf/logback-yarn.xml org.apache.flink.yarn.cli.FlinkYarnSessionCli -j /home/hadoop/ym/flink-1.1.3/lib/flink-dist_2.10-1.1.3.jar -n 3 -jm 1024 -nm 1024 -st - ** YarnApplicationMasterRunner **
/home/hadoop/ym/jdk1.8.0_101/bin/java -Xmx424M -Dlog.file=/home/hadoop/ym/hadoop-2.7.1/hadoop/nm/application_1480493133223_0009/container_1480493133223_0009_01_000001/jobmanager.log -Dlogback.configurationFile=file:logback.xml -Dlog4j.configuration=file:log4j.properties org.apache.flink.yarn.YarnApplicationMasterRunner - **个YarnTaskManager **
/home/hadoop/ym/jdk1.8.0_101/bin/java -Xms424m -Xmx424m -XX:MaxDirectMemorySize=424m -Dlog.file=/home/hadoop/ym/hadoop-2.7.1/hadoop/nm/application_1480493133223_0009/container_1480493133223_0009_01_000003/taskmanager.log -Dlogback.configurationFile=file:./logback.xml -Dlog4j.configuration=file:./log4j.properties org.apache.flink.yarn.YarnTaskManager --configDir .
后面我会从源代码的角度看下启动流程