SparkInterpreterLauncher extends StandardInterpreterLauncher { private String getEnv(String envName) { String env = properties.getProperty(envName); if (env == null) { env = System.getenv(envName); } return env; } SparkInterpreterLauncher(ZeppelinConfiguration zConf, RecoveryStorage recoveryStorage); @Override Map<String, String> buildEnvFromProperties(InterpreterLaunchContext context); static final String SPARK_MASTER_KEY; }
@Test public void testConnectTimeOut() throws IOException { ZeppelinConfiguration zConf = ZeppelinConfiguration.create(); SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null); Properties properties = new Properties(); properties.setProperty("SPARK_HOME", sparkHome); properties.setProperty( ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT.getVarName(), "10000"); InterpreterOption option = new InterpreterOption(); option.setUserImpersonate(true); InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "user1", "intpGroupId", "groupId", "groupName", "name", 0, "host"); InterpreterClient client = launcher.launch(context); assertTrue( client instanceof RemoteInterpreterManagedProcess); RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client; assertEquals("name", interpreterProcess.getInterpreterSettingName()); assertEquals(zeppelinHome + "/interpreter/groupName", interpreterProcess.getInterpreterDir()); assertEquals(zeppelinHome + "/local-repo/groupId", interpreterProcess.getLocalRepoDir()); assertEquals(10000, interpreterProcess.getConnectTimeout()); assertEquals(zConf.getInterpreterRemoteRunnerPath(), interpreterProcess.getInterpreterRunner()); assertTrue(interpreterProcess.getEnv().size() >= 2); assertEquals(true, interpreterProcess.isUserImpersonated()); }
@Test public void testLocalMode() throws IOException { ZeppelinConfiguration zConf = ZeppelinConfiguration.create(); SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null); Properties properties = new Properties(); properties.setProperty("SPARK_HOME", sparkHome); properties.setProperty("ENV_1", ""); properties.setProperty("property_1", "value_1"); properties.setProperty("spark.master", "local[*]"); properties.setProperty("spark.files", "file_1"); properties.setProperty("spark.jars", "jar_1"); InterpreterOption option = new InterpreterOption(); InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "user1", "intpGroupId", "groupId", "spark", "spark", 0, "host"); InterpreterClient client = launcher.launch(context); assertTrue( client instanceof RemoteInterpreterManagedProcess); RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client; assertEquals("spark", interpreterProcess.getInterpreterSettingName()); assertTrue(interpreterProcess.getInterpreterDir().endsWith("/interpreter/spark")); assertTrue(interpreterProcess.getLocalRepoDir().endsWith("/local-repo/groupId")); assertEquals(zConf.getInterpreterRemoteRunnerPath(), interpreterProcess.getInterpreterRunner()); assertTrue(interpreterProcess.getEnv().size() >= 2); assertEquals(sparkHome, interpreterProcess.getEnv().get("SPARK_HOME")); assertFalse(interpreterProcess.getEnv().containsKey("ENV_1")); assertEquals(InterpreterLauncher.escapeSpecialCharacter(" --conf spark.files=file_1" + " --conf spark.jars=jar_1 --conf spark.master=local[*]"), interpreterProcess.getEnv().get("ZEPPELIN_SPARK_CONF")); }
@Test public void testYarnClientMode_1() throws IOException { ZeppelinConfiguration zConf = ZeppelinConfiguration.create(); SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null); Properties properties = new Properties(); properties.setProperty("SPARK_HOME", sparkHome); properties.setProperty("property_1", "value_1"); properties.setProperty("spark.master", "yarn-client"); properties.setProperty("spark.files", "file_1"); properties.setProperty("spark.jars", "jar_1"); InterpreterOption option = new InterpreterOption(); InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "user1", "intpGroupId", "groupId", "spark", "spark", 0, "host"); InterpreterClient client = launcher.launch(context); assertTrue( client instanceof RemoteInterpreterManagedProcess); RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client; assertEquals("spark", interpreterProcess.getInterpreterSettingName()); assertTrue(interpreterProcess.getInterpreterDir().endsWith("/interpreter/spark")); assertTrue(interpreterProcess.getLocalRepoDir().endsWith("/local-repo/groupId")); assertEquals(zConf.getInterpreterRemoteRunnerPath(), interpreterProcess.getInterpreterRunner()); assertTrue(interpreterProcess.getEnv().size() >= 2); assertEquals(sparkHome, interpreterProcess.getEnv().get("SPARK_HOME")); String sparkJars = "jar_1"; String sparkrZip = sparkHome + "/R/lib/sparkr.zip#sparkr"; String sparkFiles = "file_1"; assertEquals(InterpreterLauncher.escapeSpecialCharacter(" --conf spark.yarn.dist.archives=" + sparkrZip + " --conf spark.files=" + sparkFiles + " --conf spark.jars=" + sparkJars + " --conf spark.yarn.isPython=true --conf spark.master=yarn-client"), interpreterProcess.getEnv().get("ZEPPELIN_SPARK_CONF")); }
@Test public void testYarnClientMode_2() throws IOException { ZeppelinConfiguration zConf = ZeppelinConfiguration.create(); SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null); Properties properties = new Properties(); properties.setProperty("SPARK_HOME", sparkHome); properties.setProperty("property_1", "value_1"); properties.setProperty("spark.master", "yarn"); properties.setProperty("spark.submit.deployMode", "client"); properties.setProperty("spark.files", "file_1"); properties.setProperty("spark.jars", "jar_1"); InterpreterOption option = new InterpreterOption(); InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "user1", "intpGroupId", "groupId", "spark", "spark", 0, "host"); InterpreterClient client = launcher.launch(context); assertTrue( client instanceof RemoteInterpreterManagedProcess); RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client; assertEquals("spark", interpreterProcess.getInterpreterSettingName()); assertTrue(interpreterProcess.getInterpreterDir().endsWith("/interpreter/spark")); assertTrue(interpreterProcess.getLocalRepoDir().endsWith("/local-repo/groupId")); assertEquals(zConf.getInterpreterRemoteRunnerPath(), interpreterProcess.getInterpreterRunner()); assertTrue(interpreterProcess.getEnv().size() >= 2); assertEquals(sparkHome, interpreterProcess.getEnv().get("SPARK_HOME")); String sparkJars = "jar_1"; String sparkrZip = sparkHome + "/R/lib/sparkr.zip#sparkr"; String sparkFiles = "file_1"; assertEquals(InterpreterLauncher.escapeSpecialCharacter(" --conf spark.yarn.dist.archives=" + sparkrZip + " --conf spark.files=" + sparkFiles + " --conf spark.jars=" + sparkJars + " --conf spark.submit.deployMode=client" + " --conf spark.yarn.isPython=true --conf spark.master=yarn"), interpreterProcess.getEnv().get("ZEPPELIN_SPARK_CONF")); }
@Test public void testYarnClusterMode_1() throws IOException { ZeppelinConfiguration zConf = ZeppelinConfiguration.create(); SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null); Properties properties = new Properties(); properties.setProperty("SPARK_HOME", sparkHome); properties.setProperty("property_1", "value_1"); properties.setProperty("spark.master", "yarn-cluster"); properties.setProperty("spark.files", "file_1"); properties.setProperty("spark.jars", "jar_1"); InterpreterOption option = new InterpreterOption(); InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "user1", "intpGroupId", "groupId", "spark", "spark", 0, "host"); InterpreterClient client = launcher.launch(context); assertTrue( client instanceof RemoteInterpreterManagedProcess); RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client; assertEquals("spark", interpreterProcess.getInterpreterSettingName()); assertTrue(interpreterProcess.getInterpreterDir().endsWith("/interpreter/spark")); assertTrue(interpreterProcess.getLocalRepoDir().endsWith("/local-repo/groupId")); assertEquals(zConf.getInterpreterRemoteRunnerPath(), interpreterProcess.getInterpreterRunner()); assertTrue(interpreterProcess.getEnv().size() >= 3); assertEquals(sparkHome, interpreterProcess.getEnv().get("SPARK_HOME")); assertEquals("true", interpreterProcess.getEnv().get("ZEPPELIN_SPARK_YARN_CLUSTER")); String sparkJars = "jar_1," + zeppelinHome + "/interpreter/spark/scala-2.11/spark-scala-2.11-" + Util.getVersion() + ".jar," + zeppelinHome + "/interpreter/zeppelin-interpreter-shaded-" + Util.getVersion() + ".jar"; String sparkrZip = sparkHome + "/R/lib/sparkr.zip#sparkr"; String sparkFiles = "file_1," + zeppelinHome + "/conf/log4j_yarn_cluster.properties"; assertEquals(InterpreterLauncher.escapeSpecialCharacter(" --conf spark.yarn.dist.archives=" + sparkrZip + " --conf spark.yarn.maxAppAttempts=1" + " --conf spark.files=" + sparkFiles + " --conf spark.jars=" + sparkJars + " --conf spark.yarn.isPython=true" + " --conf spark.yarn.submit.waitAppCompletion=false" + " --conf spark.master=yarn-cluster"), interpreterProcess.getEnv().get("ZEPPELIN_SPARK_CONF")); }
@Test public void testYarnClusterMode_2() throws IOException { ZeppelinConfiguration zConf = ZeppelinConfiguration.create(); SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null); Properties properties = new Properties(); properties.setProperty("SPARK_HOME", sparkHome); properties.setProperty("property_1", "value_1"); properties.setProperty("spark.master", "yarn"); properties.setProperty("spark.submit.deployMode", "cluster"); properties.setProperty("spark.files", "file_1"); properties.setProperty("spark.jars", "jar_1"); InterpreterOption option = new InterpreterOption(); option.setUserImpersonate(true); InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "user1", "intpGroupId", "groupId", "spark", "spark", 0, "host"); Path localRepoPath = Paths.get(zConf.getInterpreterLocalRepoPath(), context.getInterpreterSettingId()); FileUtils.deleteDirectory(localRepoPath.toFile()); Files.createDirectories(localRepoPath); Files.createFile(Paths.get(localRepoPath.toAbsolutePath().toString(), "test.jar")); InterpreterClient client = launcher.launch(context); assertTrue( client instanceof RemoteInterpreterManagedProcess); RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client; assertEquals("spark", interpreterProcess.getInterpreterSettingName()); assertTrue(interpreterProcess.getInterpreterDir().endsWith("/interpreter/spark")); assertTrue(interpreterProcess.getLocalRepoDir().endsWith("/local-repo/groupId")); assertEquals(zConf.getInterpreterRemoteRunnerPath(), interpreterProcess.getInterpreterRunner()); assertTrue(interpreterProcess.getEnv().size() >= 3); assertEquals(sparkHome, interpreterProcess.getEnv().get("SPARK_HOME")); assertEquals("true", interpreterProcess.getEnv().get("ZEPPELIN_SPARK_YARN_CLUSTER")); String sparkJars = "jar_1," + Paths.get(localRepoPath.toAbsolutePath().toString(), "test.jar").toString() + "," + zeppelinHome + "/interpreter/spark/scala-2.11/spark-scala-2.11-" + Util.getVersion() + ".jar," + zeppelinHome + "/interpreter/zeppelin-interpreter-shaded-" + Util.getVersion() + ".jar"; String sparkrZip = sparkHome + "/R/lib/sparkr.zip#sparkr"; String sparkFiles = "file_1," + zeppelinHome + "/conf/log4j_yarn_cluster.properties"; assertEquals(InterpreterLauncher.escapeSpecialCharacter(" --conf spark.yarn.dist.archives=" + sparkrZip + " --conf spark.yarn.maxAppAttempts=1" + " --conf spark.files=" + sparkFiles + " --conf spark.jars=" + sparkJars + " --conf spark.submit.deployMode=cluster --conf spark.yarn.isPython=true" + " --conf spark.yarn.submit.waitAppCompletion=false --conf spark.master=yarn" + " --proxy-user user1"), interpreterProcess.getEnv().get("ZEPPELIN_SPARK_CONF")); Files.deleteIfExists(Paths.get(localRepoPath.toAbsolutePath().toString(), "test.jar")); FileUtils.deleteDirectory(localRepoPath.toFile()); }
@Test public void testYarnClusterMode_3() throws IOException { ZeppelinConfiguration zConf = ZeppelinConfiguration.create(); SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null); Properties properties = new Properties(); properties.setProperty("SPARK_HOME", sparkHome); properties.setProperty("property_1", "value_1"); properties.setProperty("spark.master", "yarn"); properties.setProperty("spark.submit.deployMode", "cluster"); properties.setProperty("spark.files", "{}"); properties.setProperty("spark.jars", "jar_1"); InterpreterOption option = new InterpreterOption(); option.setUserImpersonate(true); InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "user1", "intpGroupId", "groupId", "spark", "spark", 0, "host"); Path localRepoPath = Paths.get(zConf.getInterpreterLocalRepoPath(), context.getInterpreterSettingId()); FileUtils.deleteDirectory(localRepoPath.toFile()); Files.createDirectories(localRepoPath); InterpreterClient client = launcher.launch(context); assertTrue( client instanceof RemoteInterpreterManagedProcess); RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client; assertEquals("spark", interpreterProcess.getInterpreterSettingName()); assertTrue(interpreterProcess.getInterpreterDir().endsWith("/interpreter/spark")); assertTrue(interpreterProcess.getLocalRepoDir().endsWith("/local-repo/groupId")); assertEquals(zConf.getInterpreterRemoteRunnerPath(), interpreterProcess.getInterpreterRunner()); assertTrue(interpreterProcess.getEnv().size() >= 3); assertEquals(sparkHome, interpreterProcess.getEnv().get("SPARK_HOME")); assertEquals("true", interpreterProcess.getEnv().get("ZEPPELIN_SPARK_YARN_CLUSTER")); String sparkJars = "jar_1," + zeppelinHome + "/interpreter/spark/scala-2.11/spark-scala-2.11-" + Util.getVersion() + ".jar," + zeppelinHome + "/interpreter/zeppelin-interpreter-shaded-" + Util.getVersion() + ".jar"; String sparkrZip = sparkHome + "/R/lib/sparkr.zip#sparkr"; String sparkFiles = "{}," + zeppelinHome + "/conf/log4j_yarn_cluster.properties"; assertEquals(InterpreterLauncher.escapeSpecialCharacter(" --conf spark.yarn.dist.archives=" + sparkrZip + " --conf spark.yarn.maxAppAttempts=1" + " --conf spark.files=" + sparkFiles + " --conf spark.jars=" + sparkJars + " --conf spark.submit.deployMode=cluster --conf spark.yarn.isPython=true" + " --conf spark.yarn.submit.waitAppCompletion=false" + " --conf spark.master=yarn --proxy-user user1"), interpreterProcess.getEnv().get("ZEPPELIN_SPARK_CONF")); FileUtils.deleteDirectory(localRepoPath.toFile()); }