You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zeppelin.apache.org by zj...@apache.org on 2021/08/02 03:51:11 UTC
[zeppelin] branch master updated: [ZEPPELIN-5471] spark yarn app is
not stopped when cancel button is clicked
This is an automated email from the ASF dual-hosted git repository.
zjffdu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/zeppelin.git
The following commit(s) were added to refs/heads/master by this push:
new 2ac9b1d [ZEPPELIN-5471] spark yarn app is not stopped when cancel button is clicked
2ac9b1d is described below
commit 2ac9b1df90530e3ff97c68109c6011b92042b6f3
Author: Jeff Zhang <zj...@apache.org>
AuthorDate: Wed Jul 28 16:02:07 2021 +0800
[ZEPPELIN-5471] spark yarn app is not stopped when cancel button is clicked
### What is this PR for?
Currently, SparkSubmitInterpreter only kill the client process, but leave the spark yarn app alive when cancel method is call.
This PR would detect yarn app id from logs and associate it with the current paragraph. And in the cancel method, it would kill the associated yarn app.
### What type of PR is it?
[ Improvement ]
### Todos
* [ ] - Task
### What is the Jira issue?
* https://issues.apache.org/jira/browse/ZEPPELIN-5471
### How should this be tested?
* CI pass
### Screenshots (if appropriate)
### Questions:
* Does the licenses files need update? No
* Is there breaking changes for older versions? No
* Does this needs documentation? No
Author: Jeff Zhang <zj...@apache.org>
Closes #4188 from zjffdu/ZEPPELIN-5471 and squashes the following commits:
8875f05852 [Jeff Zhang] [ZEPPELIN-5471] spark yarn app is not stopped when cancel button is clicked
---
.github/workflows/core.yml | 2 +-
.../spark/submit/SparkSubmitInterpreter.java | 47 ++++-
zeppelin-interpreter-integration/README.md | 4 +
zeppelin-interpreter-integration/pom.xml | 7 +
.../integration/SparkSubmitIntegrationTest.java | 191 +++++++++++++++++++++
5 files changed, 247 insertions(+), 4 deletions(-)
diff --git a/.github/workflows/core.yml b/.github/workflows/core.yml
index 2787a5c..fdf28ff 100644
--- a/.github/workflows/core.yml
+++ b/.github/workflows/core.yml
@@ -216,7 +216,7 @@ jobs:
mvn install -DskipTests -DskipRat -pl zeppelin-interpreter-integration,zeppelin-web,spark-submit,spark/spark-dependencies,markdown -am -Phadoop2 -Pintegration -B
mvn clean package -pl zeppelin-plugins -amd -DskipTests -B
- name: run tests
- run: mvn test -DskipRat -pl zeppelin-interpreter-integration,zeppelin-web,spark-submit,spark/spark-dependencies,markdown -am -Phadoop2 -Pintegration -B -Dtest=ZeppelinSparkClusterTest24,SparkIntegrationTest24,ZeppelinSparkClusterTest23,SparkIntegrationTest23,ZeppelinSparkClusterTest22,SparkIntegrationTest22,ZeppelinSparkClusterTest30,SparkIntegrationTest30 -DfailIfNoTests=false
+ run: mvn test -DskipRat -pl zeppelin-interpreter-integration,zeppelin-web,spark-submit,spark/spark-dependencies,markdown -am -Phadoop2 -Pintegration -B -Dtest=ZeppelinSparkClusterTest24,SparkSubmitIntegrationTest,SparkIntegrationTest24,ZeppelinSparkClusterTest23,SparkIntegrationTest23,ZeppelinSparkClusterTest22,SparkIntegrationTest22,ZeppelinSparkClusterTest30,SparkIntegrationTest30 -DfailIfNoTests=false
jdbcIntegrationTest-and-unit-test-of-Spark-2-4-with-Scala-2-11:
runs-on: ubuntu-20.04
steps:
diff --git a/spark-submit/src/main/java/org/apache/zeppelin/spark/submit/SparkSubmitInterpreter.java b/spark-submit/src/main/java/org/apache/zeppelin/spark/submit/SparkSubmitInterpreter.java
index 7dce340..61a1c95 100644
--- a/spark-submit/src/main/java/org/apache/zeppelin/spark/submit/SparkSubmitInterpreter.java
+++ b/spark-submit/src/main/java/org/apache/zeppelin/spark/submit/SparkSubmitInterpreter.java
@@ -18,6 +18,8 @@
package org.apache.zeppelin.spark.submit;
+import org.apache.commons.exec.environment.EnvironmentUtils;
+import org.apache.commons.lang3.StringUtils;
import org.apache.zeppelin.interpreter.InterpreterContext;
import org.apache.zeppelin.interpreter.InterpreterOutput;
import org.apache.zeppelin.interpreter.InterpreterOutputListener;
@@ -27,8 +29,11 @@ import org.apache.zeppelin.shell.ShellInterpreter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
/**
@@ -42,6 +47,9 @@ public class SparkSubmitInterpreter extends ShellInterpreter {
private String sparkHome;
+ // paragraphId --> yarnAppId
+ private ConcurrentMap<String, String> yarnAppIdMap = new ConcurrentHashMap();
+
public SparkSubmitInterpreter(Properties property) {
super(property);
// Set time to be max integer so that the shell process won't timeout.
@@ -52,16 +60,37 @@ public class SparkSubmitInterpreter extends ShellInterpreter {
@Override
public InterpreterResult internalInterpret(String cmd, InterpreterContext context) {
+ if (StringUtils.isBlank(cmd)) {
+ return new InterpreterResult(InterpreterResult.Code.SUCCESS);
+ }
String sparkSubmitCommand = sparkHome + "/bin/spark-submit " + cmd.trim();
LOGGER.info("Run spark command: " + sparkSubmitCommand);
context.out.addInterpreterOutListener(new SparkSubmitOutputListener(context));
- return super.internalInterpret(sparkSubmitCommand, context);
+ InterpreterResult result = super.internalInterpret(sparkSubmitCommand, context);
+ yarnAppIdMap.remove(context.getParagraphId());
+ return result;
+ }
+
+ @Override
+ public void cancel(InterpreterContext context) {
+ super.cancel(context);
+ String yarnAppId = yarnAppIdMap.remove(context.getParagraphId());
+ if (StringUtils.isNotBlank(yarnAppId)) {
+ try {
+ LOGGER.info("Try to kill yarn app: {} of code: {}", yarnAppId, context.getParagraphText());
+ Runtime.getRuntime().exec(new String[] {"yarn", "application", "-kill", yarnAppId});
+ } catch (IOException e) {
+ LOGGER.warn("Fail to kill yarn app, please check whether yarn command is on your PATH", e);
+ }
+ } else {
+ LOGGER.warn("No yarn app associated with code: {}", context.getParagraphText());
+ }
}
/**
* InterpreterOutputListener which extract spark ui link from logs.
*/
- private static class SparkSubmitOutputListener implements InterpreterOutputListener {
+ private class SparkSubmitOutputListener implements InterpreterOutputListener {
private InterpreterContext context;
private boolean isSparkUrlSent = false;
@@ -78,6 +107,7 @@ public class SparkSubmitInterpreter extends ShellInterpreter {
@Override
public void onAppend(int index, InterpreterResultMessageOutput out, byte[] line) {
String text = new String(line);
+ LOGGER.debug("Output: {}", text);
if (isSparkUrlSent) {
return;
}
@@ -95,7 +125,8 @@ public class SparkSubmitInterpreter extends ShellInterpreter {
private void buildSparkUIInfo(String log, InterpreterContext context) {
int pos = log.lastIndexOf(" ");
if (pos != -1) {
- String sparkUI = log.substring(pos + 1);
+ String sparkUI = log.substring(pos + 1).trim();
+ sparkUI = sparkUI.substring(0, sparkUI.length() - 1);
Map<String, String> infos = new java.util.HashMap<String, String>();
infos.put("jobUrl", sparkUI);
infos.put("label", "Spark UI");
@@ -103,6 +134,16 @@ public class SparkSubmitInterpreter extends ShellInterpreter {
infos.put("noteId", context.getNoteId());
infos.put("paraId", context.getParagraphId());
context.getIntpEventClient().onParaInfosReceived(infos);
+
+ // extract yarn appId
+ int lastIndexOfSlash = sparkUI.lastIndexOf("/");
+ if (lastIndexOfSlash != -1) {
+ String yarnAppId = sparkUI.substring(lastIndexOfSlash + 1);
+ LOGGER.info("Detected yarn app: {}", yarnAppId);
+ SparkSubmitInterpreter.this.yarnAppIdMap.put(context.getParagraphId(), yarnAppId);
+ } else {
+ LOGGER.warn("Might be an invalid spark URL: " + sparkUI);
+ }
} else {
LOGGER.error("Unable to extract spark url from this log: " + log);
}
diff --git a/zeppelin-interpreter-integration/README.md b/zeppelin-interpreter-integration/README.md
new file mode 100644
index 0000000..bf235ff
--- /dev/null
+++ b/zeppelin-interpreter-integration/README.md
@@ -0,0 +1,4 @@
+## How to run Zeppelin integration tests
+
+If you have hadoop installed on your machine, please make sure to unset hadoop related enviromnets:
+* HADOOP_CONF_DIR
\ No newline at end of file
diff --git a/zeppelin-interpreter-integration/pom.xml b/zeppelin-interpreter-integration/pom.xml
index 2237b41..9e1c913 100644
--- a/zeppelin-interpreter-integration/pom.xml
+++ b/zeppelin-interpreter-integration/pom.xml
@@ -151,6 +151,13 @@
</dependency>
<dependency>
+ <groupId>net.jodah</groupId>
+ <artifactId>concurrentunit</artifactId>
+ <version>0.4.4</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-all</artifactId>
<scope>test</scope>
diff --git a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/SparkSubmitIntegrationTest.java b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/SparkSubmitIntegrationTest.java
new file mode 100644
index 0000000..7575aa0
--- /dev/null
+++ b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/SparkSubmitIntegrationTest.java
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.zeppelin.integration;
+
+import net.jodah.concurrentunit.Waiter;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsResponse;
+import org.apache.hadoop.yarn.api.records.ApplicationReport;
+import org.apache.hadoop.yarn.api.records.YarnApplicationState;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.zeppelin.interpreter.ExecutionContext;
+import org.apache.zeppelin.interpreter.Interpreter;
+import org.apache.zeppelin.interpreter.InterpreterContext;
+import org.apache.zeppelin.interpreter.InterpreterException;
+import org.apache.zeppelin.interpreter.InterpreterFactory;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.apache.zeppelin.interpreter.InterpreterSetting;
+import org.apache.zeppelin.interpreter.InterpreterSettingManager;
+import org.apache.zeppelin.interpreter.integration.DownloadUtils;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class SparkSubmitIntegrationTest {
+
+ private static Logger LOGGER = LoggerFactory.getLogger(SparkSubmitIntegrationTest.class);
+
+ private static MiniHadoopCluster hadoopCluster;
+ private static MiniZeppelin zeppelin;
+ private static InterpreterFactory interpreterFactory;
+ private static InterpreterSettingManager interpreterSettingManager;
+
+ private static String sparkHome;
+
+ @BeforeClass
+ public static void setUp() throws IOException {
+ String sparkVersion = "2.4.7";
+ String hadoopVersion = "2.7";
+ LOGGER.info("Testing Spark Version: " + sparkVersion);
+ LOGGER.info("Testing Hadoop Version: " + hadoopVersion);
+ sparkHome = DownloadUtils.downloadSpark(sparkVersion, hadoopVersion);
+
+ hadoopCluster = new MiniHadoopCluster();
+ hadoopCluster.start();
+
+ zeppelin = new MiniZeppelin();
+ zeppelin.start(SparkIntegrationTest.class);
+ interpreterFactory = zeppelin.getInterpreterFactory();
+ interpreterSettingManager = zeppelin.getInterpreterSettingManager();
+
+ InterpreterSetting sparkSubmitInterpreterSetting =
+ interpreterSettingManager.getInterpreterSettingByName("spark-submit");
+ sparkSubmitInterpreterSetting.setProperty("SPARK_HOME", sparkHome);
+ sparkSubmitInterpreterSetting.setProperty("HADOOP_CONF_DIR", hadoopCluster.getConfigPath());
+ sparkSubmitInterpreterSetting.setProperty("YARN_CONF_DIR", hadoopCluster.getConfigPath());
+ }
+
+ @AfterClass
+ public static void tearDown() throws IOException {
+ if (zeppelin != null) {
+ zeppelin.stop();
+ }
+ if (hadoopCluster != null) {
+ hadoopCluster.stop();
+ }
+ }
+
+ @Test
+ public void testLocalMode() throws InterpreterException, YarnException {
+ try {
+ // test SparkSubmitInterpreterSetting
+ Interpreter sparkSubmitInterpreter = interpreterFactory.getInterpreter("spark-submit", new ExecutionContext("user1", "note1", "test"));
+
+ InterpreterContext context = new InterpreterContext.Builder().setNoteId("note1").setParagraphId("paragraph_1").build();
+ InterpreterResult interpreterResult =
+ sparkSubmitInterpreter.interpret("--master yarn-cluster --class org.apache.spark.examples.SparkPi " +
+ sparkHome + "/examples/jars/spark-examples_2.11-2.4.7.jar", context);
+ assertEquals(interpreterResult.toString(), InterpreterResult.Code.SUCCESS, interpreterResult.code());
+
+ // no yarn application launched
+ GetApplicationsRequest request = GetApplicationsRequest.newInstance(EnumSet.of(YarnApplicationState.RUNNING));
+ GetApplicationsResponse response = hadoopCluster.getYarnCluster().getResourceManager().getClientRMService().getApplications(request);
+ assertEquals(0, response.getApplicationList().size());
+ } finally {
+ interpreterSettingManager.close();
+ }
+ }
+
+ @Test
+ public void testYarnMode() throws InterpreterException, YarnException {
+ try {
+ // test SparkSubmitInterpreterSetting
+ Interpreter sparkSubmitInterpreter = interpreterFactory.getInterpreter("spark-submit", new ExecutionContext("user1", "note1", "test"));
+
+ InterpreterContext context = new InterpreterContext.Builder().setNoteId("note1").setParagraphId("paragraph_1").build();
+ String yarnAppName = "yarn_example";
+ InterpreterResult interpreterResult =
+ sparkSubmitInterpreter.interpret("--master yarn-cluster --class org.apache.spark.examples.SparkPi " +
+ "--conf spark.app.name=" + yarnAppName + " --conf spark.driver.memory=512m " +
+ "--conf spark.executor.memory=512m " +
+ sparkHome + "/examples/jars/spark-examples_2.11-2.4.7.jar", context);
+ assertEquals(interpreterResult.toString(), InterpreterResult.Code.SUCCESS, interpreterResult.code());
+
+ GetApplicationsRequest request = GetApplicationsRequest.newInstance(EnumSet.of(YarnApplicationState.FINISHED));
+ GetApplicationsResponse response = hadoopCluster.getYarnCluster().getResourceManager().getClientRMService().getApplications(request);
+ assertTrue(response.getApplicationList().size() >= 1);
+
+ List<ApplicationReport> apps = response.getApplicationList().stream()
+ .filter(app -> app.getName().equals(yarnAppName))
+ .collect(Collectors.toList());
+ assertEquals(1, apps.size());
+ } finally {
+ interpreterSettingManager.close();
+ }
+ }
+
+ @Test
+ public void testCancelSparkYarnApp() throws InterpreterException, YarnException, TimeoutException, InterruptedException {
+ try {
+ // test SparkSubmitInterpreterSetting
+ Interpreter sparkSubmitInterpreter = interpreterFactory.getInterpreter("spark-submit", new ExecutionContext("user1", "note1", "test"));
+ InterpreterContext context = new InterpreterContext.Builder().setNoteId("note1").setParagraphId("paragraph_1").build();
+
+ final Waiter waiter = new Waiter();
+ Thread thread = new Thread() {
+ @Override
+ public void run() {
+ try {
+ String yarnAppName = "yarn_cancel_example";
+ InterpreterResult interpreterResult =
+ sparkSubmitInterpreter.interpret("--master yarn-cluster --class org.apache.spark.examples.SparkPi " +
+ "--conf spark.app.name=" + yarnAppName + " --conf spark.driver.memory=512m " +
+ "--conf spark.executor.memory=512m " +
+ sparkHome + "/examples/jars/spark-examples_2.11-2.4.7.jar", context);
+ assertEquals(interpreterResult.toString(), InterpreterResult.Code.INCOMPLETE, interpreterResult.code());
+ assertTrue(interpreterResult.toString(), interpreterResult.toString().contains("Paragraph received a SIGTERM"));
+ } catch (InterpreterException e) {
+ waiter.fail("Should not throw exception\n" + ExceptionUtils.getStackTrace(e));
+ }
+ waiter.resume();
+ }
+ };
+ thread.start();
+
+ long start = System.currentTimeMillis();
+ long threshold = 120 * 1000;
+ while ((System.currentTimeMillis() - start) < threshold) {
+ GetApplicationsRequest request = GetApplicationsRequest.newInstance(EnumSet.of(YarnApplicationState.RUNNING));
+ GetApplicationsResponse response = hadoopCluster.getYarnCluster().getResourceManager().getClientRMService().getApplications(request);
+ if (response.getApplicationList().size() >= 1) {
+ break;
+ }
+ Thread.sleep(5 * 1000);
+ }
+
+ sparkSubmitInterpreter.cancel(context);
+ waiter.await(10000);
+
+ } finally {
+ interpreterSettingManager.close();
+ }
+ }
+}