@Test public void testGenerateSparkCmd() { KylinConfig kylinConfig = getTestConfig(); final String kylinHome = System.getProperty("KYLIN_HOME"); try { System.setProperty("KYLIN_HOME", "/kylin"); NSparkExecutable sparkExecutable = new NSparkExecutable(); sparkExecutable.setProject("default"); String hadoopConf = System.getProperty("KYLIN_HOME") + "/hadoop"; String kylinJobJar = System.getProperty("KYLIN_HOME") + "/lib/job.jar"; String appArgs = "/tmp/output"; String cmd = sparkExecutable.generateSparkCmd(kylinConfig, hadoopConf, kylinJobJar, kylinJobJar, appArgs); Assert.assertNotNull(cmd); Assert.assertTrue(cmd.contains("export HADOOP_CONF_DIR")); Assert.assertTrue(cmd.contains("spark-submit")); Assert.assertTrue(cmd.contains("spark.executor.extraClassPath=job.jar")); Assert.assertTrue(cmd.contains("spark.driver.log4j.appender.hdfs.File=")); } finally { if (StringUtils.isEmpty(kylinHome)) { System.clearProperty("KYLIN_HOME"); } else { System.setProperty("KYLIN_HOME", kylinHome); } } }
protected String generateSparkCmd(KylinConfig config, String hadoopConf, String jars, String kylinJobJar, String appArgs) { StringBuilder sb = new StringBuilder(); sb.append( "export HADOOP_CONF_DIR=%s && %s/bin/spark-submit --class org.apache.kylin.engine.spark.application.SparkEntry "); Map<String, String> sparkConfs = getSparkConfigOverride(config); for (Entry<String, String> entry : sparkConfs.entrySet()) { appendSparkConf(sb, entry.getKey(), entry.getValue()); } appendSparkConf(sb, "spark.executor.extraClassPath", Paths.get(kylinJobJar).getFileName().toString()); appendSparkConf(sb, "spark.driver.extraClassPath", kylinJobJar); if (sparkConfs.containsKey("spark.sql.hive.metastore.jars")) { jars = jars + "," + sparkConfs.get("spark.sql.hive.metastore.jars"); } sb.append("--name job_step_%s "); sb.append("--jars %s %s %s"); String cmd = String.format(Locale.ROOT, sb.toString(), hadoopConf, KylinConfig.getSparkHome(), getId(), jars, kylinJobJar, appArgs); logger.info("spark submit cmd: {}", cmd); return cmd; }
NSparkExecutable extends AbstractExecutable { protected String generateSparkCmd(KylinConfig config, String hadoopConf, String jars, String kylinJobJar, String appArgs) { StringBuilder sb = new StringBuilder(); sb.append( "export HADOOP_CONF_DIR=%s && %s/bin/spark-submit --class org.apache.kylin.engine.spark.application.SparkEntry "); Map<String, String> sparkConfs = getSparkConfigOverride(config); for (Entry<String, String> entry : sparkConfs.entrySet()) { appendSparkConf(sb, entry.getKey(), entry.getValue()); } appendSparkConf(sb, "spark.executor.extraClassPath", Paths.get(kylinJobJar).getFileName().toString()); appendSparkConf(sb, "spark.driver.extraClassPath", kylinJobJar); if (sparkConfs.containsKey("spark.sql.hive.metastore.jars")) { jars = jars + "," + sparkConfs.get("spark.sql.hive.metastore.jars"); } sb.append("--name job_step_%s "); sb.append("--jars %s %s %s"); String cmd = String.format(Locale.ROOT, sb.toString(), hadoopConf, KylinConfig.getSparkHome(), getId(), jars, kylinJobJar, appArgs); logger.info("spark submit cmd: {}", cmd); return cmd; } }
NSparkExecutable extends AbstractExecutable { protected String generateSparkCmd(KylinConfig config, String hadoopConf, String jars, String kylinJobJar, String appArgs) { StringBuilder sb = new StringBuilder(); sb.append( "export HADOOP_CONF_DIR=%s && %s/bin/spark-submit --class org.apache.kylin.engine.spark.application.SparkEntry "); Map<String, String> sparkConfs = getSparkConfigOverride(config); for (Entry<String, String> entry : sparkConfs.entrySet()) { appendSparkConf(sb, entry.getKey(), entry.getValue()); } appendSparkConf(sb, "spark.executor.extraClassPath", Paths.get(kylinJobJar).getFileName().toString()); appendSparkConf(sb, "spark.driver.extraClassPath", kylinJobJar); if (sparkConfs.containsKey("spark.sql.hive.metastore.jars")) { jars = jars + "," + sparkConfs.get("spark.sql.hive.metastore.jars"); } sb.append("--name job_step_%s "); sb.append("--jars %s %s %s"); String cmd = String.format(Locale.ROOT, sb.toString(), hadoopConf, KylinConfig.getSparkHome(), getId(), jars, kylinJobJar, appArgs); logger.info("spark submit cmd: {}", cmd); return cmd; }  }
NSparkExecutable extends AbstractExecutable { protected String generateSparkCmd(KylinConfig config, String hadoopConf, String jars, String kylinJobJar, String appArgs) { StringBuilder sb = new StringBuilder(); sb.append( "export HADOOP_CONF_DIR=%s && %s/bin/spark-submit --class org.apache.kylin.engine.spark.application.SparkEntry "); Map<String, String> sparkConfs = getSparkConfigOverride(config); for (Entry<String, String> entry : sparkConfs.entrySet()) { appendSparkConf(sb, entry.getKey(), entry.getValue()); } appendSparkConf(sb, "spark.executor.extraClassPath", Paths.get(kylinJobJar).getFileName().toString()); appendSparkConf(sb, "spark.driver.extraClassPath", kylinJobJar); if (sparkConfs.containsKey("spark.sql.hive.metastore.jars")) { jars = jars + "," + sparkConfs.get("spark.sql.hive.metastore.jars"); } sb.append("--name job_step_%s "); sb.append("--jars %s %s %s"); String cmd = String.format(Locale.ROOT, sb.toString(), hadoopConf, KylinConfig.getSparkHome(), getId(), jars, kylinJobJar, appArgs); logger.info("spark submit cmd: {}", cmd); return cmd; }  String getSparkSubmitClassName(); String getJars(); String getDistMetaUrl(); boolean needMergeMetadata(); }
NSparkExecutable extends AbstractExecutable { protected String generateSparkCmd(KylinConfig config, String hadoopConf, String jars, String kylinJobJar, String appArgs) { StringBuilder sb = new StringBuilder(); sb.append( "export HADOOP_CONF_DIR=%s && %s/bin/spark-submit --class org.apache.kylin.engine.spark.application.SparkEntry "); Map<String, String> sparkConfs = getSparkConfigOverride(config); for (Entry<String, String> entry : sparkConfs.entrySet()) { appendSparkConf(sb, entry.getKey(), entry.getValue()); } appendSparkConf(sb, "spark.executor.extraClassPath", Paths.get(kylinJobJar).getFileName().toString()); appendSparkConf(sb, "spark.driver.extraClassPath", kylinJobJar); if (sparkConfs.containsKey("spark.sql.hive.metastore.jars")) { jars = jars + "," + sparkConfs.get("spark.sql.hive.metastore.jars"); } sb.append("--name job_step_%s "); sb.append("--jars %s %s %s"); String cmd = String.format(Locale.ROOT, sb.toString(), hadoopConf, KylinConfig.getSparkHome(), getId(), jars, kylinJobJar, appArgs); logger.info("spark submit cmd: {}", cmd); return cmd; }  String getSparkSubmitClassName(); String getJars(); String getDistMetaUrl(); boolean needMergeMetadata();  }