[GitHub] [zeppelin] zjffdu commented on a change in pull request #4098: [ZEPPELIN-5320]. Support yarn application mode for flink interpreter

[GitBox <gi...@apache.org>]

zjffdu commented on a change in pull request #4098:
URL: https://github.com/apache/zeppelin/pull/4098#discussion_r615960200



##########
File path: flink/flink1.12-shims/src/main/java/org/apache/zeppelin/flink/Flink112Shims.java
##########
@@ -173,13 +173,26 @@ public Object createCatalogManager(Object config) {
 
   @Override
   public String getPyFlinkPythonPath(Properties properties) throws IOException {
+    if ("yarn_application".equalsIgnoreCase(properties.getProperty("flink.execution.mode"))) {
+      // for yarn application mode, FLINK_HOME is container working directory
+      String flinkHome = new File(".").getAbsolutePath();
+      List<File> depFiles = null;
+      depFiles = Arrays.asList(new File(flinkHome + "/lib/python").listFiles());
+      StringBuilder builder = new StringBuilder();
+      for (File file : depFiles) {
+        LOGGER.info("Adding extracted file {} to PYTHONPATH", file.getAbsolutePath());
+        builder.append(file.getAbsolutePath() + ":");
+      }
+      return builder.toString();

Review comment:
       Good point, Fixed

##########
File path: flink/interpreter/src/main/java/org/apache/zeppelin/flink/HadoopUtils.java
##########
@@ -55,6 +65,18 @@ public static String getYarnAppTrackingUrl(ClusterClient clusterClient) throws I
     return yarnClient.getApplicationReport(yarnAppId).getTrackingUrl();
   }
 
+  public static int getFlinkRestPort(String yarnAppId) throws IOException, YarnException {
+    YarnClient yarnClient = YarnClient.createYarnClient();
+    YarnConfiguration yarnConf = new YarnConfiguration();
+    // disable timeline service as we only query yarn app here.
+    // Otherwise we may hit this kind of ERROR:
+    // java.lang.ClassNotFoundException: com.sun.jersey.api.client.config.ClientConfig
+    yarnConf.set("yarn.timeline-service.enabled", "false");
+    yarnClient.init(yarnConf);
+    yarnClient.start();
+    return yarnClient.getApplicationReport(ConverterUtils.toApplicationId(yarnAppId)).getRpcPort();

Review comment:
       Good point, fixed




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org