Flink on yarn运行python程序

Flink on yarn运行python程序

Flink本地local模式运行python程序

参考官网写了python程序Adder.py(后面贴上代码)，在本地local模式用./bin/pyflink.sh ./Adder.p启动是可以的

➜  flink-1.7.1 ./bin/pyflink.sh ./Adder.py
Starting execution of program
Program execution finished
Job with JobID 8c7ab33a382c279b66089d43693fde52 has finished.
Job Runtime: 679 ms

jobManager web ui上查看是有JobID 为8c7ab33a382c279b66089d43693fde52的任务的。

可以看到本地local模式是可以运行python程序的。

Flink on yarn运行python程序

在服务器6.34上执行（6.34上有单机的flink客户端，可以连接yarn，向yarn提交flink程序）
iknow@search-uda-6-34:~/wangbin/flink-1.7.2 $ ./bin/pyflink.sh ./Adder.py

默认会起动到application_1550579025929_61820上，指定了yid参数也会默认跑在application_1550579025929_61820上

决定换一台机器试试，到0-107上，这里省略拷贝hadoop和flink环境的过程，
启动一个yarn session：

./bin/yarn-session.sh -qu core -nm yarn-session-core -n 2 -jm 1024m -tm 2096m

启动日志如下：

iknow@search-uda-0-107:~/wangbin/flink/flink-1.7.2 $ ./bin/yarn-session.sh -qu core -nm yarn-session-core -n 2 -jm 1024m -tm 2096m
2019-03-01 19:53:31,148 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.address, localhost
2019-03-01 19:53:31,150 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.port, 6123
2019-03-01 19:53:31,151 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.heap.size, 1024m
2019-03-01 19:53:31,151 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.heap.size, 1024m
2019-03-01 19:53:31,151 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.numberOfTaskSlots, 1
2019-03-01 19:53:31,152 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: parallelism.default, 1
2019-03-01 19:53:31,153 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: rest.port, 8081
2019-03-01 19:53:31,815 WARN  org.apache.hadoop.util.NativeCodeLoader                       - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2019-03-01 19:53:31,978 WARN  org.apache.hadoop.conf.Configuration                          - /home/iknow/hadoop-client/etc/hadoop/core-site.xml:an attempt to override final parameter: fs.defaultFS;  Ignoring.
2019-03-01 19:53:31,980 WARN  org.apache.hadoop.conf.Configuration                          - /home/iknow/hadoop-client/etc/hadoop/hdfs-site.xml:an attempt to override final parameter: dfs.datanode.data.dir;  Ignoring.
2019-03-01 19:53:31,981 WARN  org.apache.hadoop.conf.Configuration                          - /home/iknow/hadoop-client/etc/hadoop/hdfs-site.xml:an attempt to override final parameter: dfs.namenode.name.dir;  Ignoring.
2019-03-01 19:53:31,986 INFO  org.apache.flink.runtime.security.modules.HadoopModule        - Hadoop user set to iknow (auth:SIMPLE)
2019-03-01 19:53:32,451 INFO  org.apache.hadoop.yarn.client.AHSProxy                        - Connecting to Application History server at data-hadoop-112-16.bjrs.zybang.com/192.168.112.16:10200
2019-03-01 19:53:32,479 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - The argument n is deprecated in will be ignored.
2019-03-01 19:53:32,542 INFO  org.apache.hadoop.yarn.client.RequestHedgingRMFailoverProxyProvider  - Looking for the active RM in [rm1, rm2]...
2019-03-01 19:53:32,662 INFO  org.apache.hadoop.yarn.client.RequestHedgingRMFailoverProxyProvider  - Found active RM [rm1]
2019-03-01 19:53:32,720 WARN  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The JobManager or TaskManager memory is below the smallest possible YARN Container size. The value of 'yarn.scheduler.minimum-allocation-mb' is '2048'. Please increase the memory size.YARN will allocate the smaller containers but the scheduler will account for the minimum-allocation-mb, maybe not all instances you requested will start.
2019-03-01 19:53:32,720 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Cluster specification: ClusterSpecification{masterMemoryMB=2048, taskManagerMemoryMB=2096, numberTaskManagers=2, slotsPerTaskManager=1}
2019-03-01 19:53:33,265 WARN  org.apache.hadoop.hdfs.shortcircuit.DomainSocketFactory       - The short-circuit local reads feature cannot be used because libhadoop cannot be loaded.
2019-03-01 19:53:33,274 WARN  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The configuration directory ('/home/iknow/wangbin/flink/flink-1.7.2/conf') contains both LOG4J and Logback configuration files. Please delete or rename one of them.
2019-03-01 19:53:42,592 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Submitting application master application_1550579025929_89782
2019-03-01 19:53:42,849 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1550579025929_89782
2019-03-01 19:53:42,849 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Waiting for the cluster to be allocated
2019-03-01 19:53:42,853 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Deploying cluster, current state ACCEPTED
2019-03-01 19:53:50,718 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - YARN application has been deployed successfully.
2019-03-01 19:53:51,320 INFO  org.apache.flink.runtime.rest.RestClient                      - Rest client endpoint started.
Flink JobManager is now running on search-as-107-45.bjcq.zybang.com:42915 with leader id 00000000-0000-0000-0000-000000000000.
JobManager Web Interface: http://search-as-107-45.bjcq.zybang.com:42915

yarn上是有这个yarn session的

点击ApplicationMaster进去JobManager的web ui，启动一个python程序
./bin/pyflink.sh Adder.py程序起动到了ID为application_1550579025929_89782 的APPID上

JobID为5612d8a9b68a72633a7e2138df4537ba。

再启动一个yarn session
./bin/yarn-session.sh -qu emergency -nm yarn-session-emergency -n 4 -jm 1024m -tm 2096m

在yarn上看到有ID为application_1550579025929_89810的yarn session

再用./bin/pyflink.sh脚本运行python程序

查看JobManager 的web ui上看到是有899b20e2cde7b4ef29afc5b90e56325d的Job ID的。

也就是./bin/pyflink.sh脚本会将python程序运行在最新启动的yarn session上。

再启动一个yarn session

./bin/yarn-session.sh -qu emergency -nm yarn-session-emergency-2 -n 2 -jm 2048m -tm 4096m

Yarn上看到有ID为application_1550579025929_89846的yarn session

./bin/pyflink.sh脚本运行python脚本

到JobManager上看到有jobID为9d7d9d58ee886fd74a7ec56ff487b80d的任务。

总结

Flink on yarn运行python程序的步骤：

• 1. 开启一个yarn session
• 2. 用 ./bin/pyflink.sh脚本运行python程序，程序会跑在最新启动的yarn session上。

• 遇到的问题，如果启动多个yarn session，无法通过指定yid来运行到不同yarn session上。

参考：
https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/batch/python.html