diff --git a/conf/livy-fairscheduler.xml b/conf/livy-fairscheduler.xml new file mode 100644 index 000000000..eadcddc4a --- /dev/null +++ b/conf/livy-fairscheduler.xml @@ -0,0 +1,31 @@ + + + + + + + FIFO + 1 + 0 + + + FAIR + 1 + 0 + + diff --git a/core/src/main/scala/org/apache/livy/sessions/Kind.scala b/core/src/main/scala/org/apache/livy/sessions/Kind.scala index 0a05c8fcd..7d10def23 100644 --- a/core/src/main/scala/org/apache/livy/sessions/Kind.scala +++ b/core/src/main/scala/org/apache/livy/sessions/Kind.scala @@ -35,6 +35,8 @@ object Shared extends Kind("shared") object SQL extends Kind("sql") +object ConcurrentSQL extends Kind("concurrentSQL") + object Kind { def apply(kind: String): Kind = kind match { @@ -43,6 +45,7 @@ object Kind { case "sparkr" | "r" => SparkR case "shared" => Shared case "sql" => SQL + case "concurrentSQL" => ConcurrentSQL case other => throw new IllegalArgumentException(s"Invalid kind: $other") } } diff --git a/repl/src/main/scala/org/apache/livy/repl/SQLInterpreter.scala b/repl/src/main/scala/org/apache/livy/repl/SQLInterpreter.scala index 5a7b60638..479e97a6c 100644 --- a/repl/src/main/scala/org/apache/livy/repl/SQLInterpreter.scala +++ b/repl/src/main/scala/org/apache/livy/repl/SQLInterpreter.scala @@ -64,7 +64,8 @@ import org.apache.livy.rsc.driver.SparkEntries class SQLInterpreter( sparkConf: SparkConf, rscConf: RSCConf, - sparkEntries: SparkEntries) extends Interpreter with Logging { + sparkEntries: SparkEntries, + pool: String = "default") extends Interpreter with Logging { private implicit def formats = DefaultFormats @@ -81,6 +82,7 @@ class SQLInterpreter( override protected[repl] def execute(code: String): Interpreter.ExecuteResponse = { try { + spark.sparkContext.setLocalProperty("spark.scheduler.pool", pool) val result = spark.sql(code) val schema = parse(result.schema.json) @@ -94,6 +96,7 @@ class SQLInterpreter( case e => e } } + spark.sparkContext.setLocalProperty("spark.scheduler.pool", null) val jRows = Extraction.decompose(rows) diff --git a/repl/src/main/scala/org/apache/livy/repl/Session.scala b/repl/src/main/scala/org/apache/livy/repl/Session.scala index 4ab467c3b..13f667805 100644 --- a/repl/src/main/scala/org/apache/livy/repl/Session.scala +++ b/repl/src/main/scala/org/apache/livy/repl/Session.scala @@ -60,6 +60,9 @@ class Session( private val interpreterExecutor = ExecutionContext.fromExecutorService( Executors.newSingleThreadExecutor()) + private lazy val concurrentSQLExecutor = ExecutionContext.fromExecutorService( + Executors.newFixedThreadPool(livyConf.getInt(RSCConf.Entry.CONCURRENT_SQL_MAX))) + private val cancelExecutor = ExecutionContext.fromExecutorService( Executors.newSingleThreadExecutor()) @@ -105,6 +108,8 @@ class Session( throw new IllegalStateException("SparkInterpreter should not be lazily created.") case PySpark => PythonInterpreter(sparkConf, entries) case SparkR => SparkRInterpreter(sparkConf, entries) + case ConcurrentSQL => new SQLInterpreter(sparkConf, livyConf, entries, + livyConf.get(RSCConf.Entry.CONCURRENT_SQL_SCHEDULER_POOL)) case SQL => new SQLInterpreter(sparkConf, livyConf, entries) } interp.start() @@ -160,8 +165,14 @@ class Session( val statement = new Statement(statementId, code, StatementState.Waiting, null) _statements.synchronized { _statements(statementId) = statement } + val threadPool = if (tpe == ConcurrentSQL) { + concurrentSQLExecutor + } else { + interpreterExecutor + } + Future { - setJobGroup(tpe, statementId) + this.synchronized { setJobGroup(tpe, statementId) } statement.compareAndTransit(StatementState.Waiting, StatementState.Running) if (statement.state.get() == StatementState.Running) { @@ -171,7 +182,7 @@ class Session( statement.compareAndTransit(StatementState.Running, StatementState.Available) statement.compareAndTransit(StatementState.Cancelling, StatementState.Cancelled) statement.updateProgress(1.0) - }(interpreterExecutor) + }(threadPool) statementId } @@ -333,7 +344,7 @@ class Session( private def setJobGroup(codeType: Kind, statementId: Int): String = { val jobGroup = statementIdToJobGroup(statementId) val (cmd, tpe) = codeType match { - case Spark | SQL => + case Spark | SQL | ConcurrentSQL => // A dummy value to avoid automatic value binding in scala REPL. (s"""val _livyJobGroup$jobGroup = sc.setJobGroup("$jobGroup",""" + s""""Job group for statement $jobGroup")""", diff --git a/rsc/src/main/java/org/apache/livy/rsc/ContextLauncher.java b/rsc/src/main/java/org/apache/livy/rsc/ContextLauncher.java index 5a819d563..4e5616add 100644 --- a/rsc/src/main/java/org/apache/livy/rsc/ContextLauncher.java +++ b/rsc/src/main/java/org/apache/livy/rsc/ContextLauncher.java @@ -69,6 +69,7 @@ class ContextLauncher { private static final String SPARK_JARS_KEY = "spark.jars"; private static final String SPARK_ARCHIVES_KEY = "spark.yarn.dist.archives"; private static final String SPARK_HOME_ENV = "SPARK_HOME"; + private static final String FAIR_SCHEDULER_FILENAME = "livy-fairscheduler.xml"; static DriverProcessInfo create(RSCClientFactory factory, RSCConf conf) throws IOException { @@ -200,6 +201,21 @@ private static ChildProcess startDriver(final RSCConf conf, Promise promise) // connections for the same registered app. conf.set("spark.yarn.maxAppAttempts", "1"); + conf.set("spark.scheduler.mode", "FAIR"); + String master = conf.get(SparkLauncher.SPARK_MASTER); + String deployMode = conf.get(SparkLauncher.DEPLOY_MODE); + Boolean isCluster = (master != null && master.endsWith("-cluster")) || + (deployMode != null && deployMode.equals("cluster")); + String fairSchedulerXMLPath = System.getenv("LIVY_HOME") + + File.separator + "conf" + File.separator + FAIR_SCHEDULER_FILENAME; + if (System.getenv("LIVY_HOME") != null) { + if (isCluster) { + conf.set("spark.scheduler.allocation.file", FAIR_SCHEDULER_FILENAME); + } else { + conf.set("spark.scheduler.allocation.file", fairSchedulerXMLPath); + } + } + // Let the launcher go away when launcher in yarn cluster mode. This avoids keeping lots // of "small" Java processes lingering on the Livy server node. conf.set("spark.yarn.submit.waitAppCompletion", "false"); @@ -239,6 +255,9 @@ public void run() { } else { final SparkLauncher launcher = new SparkLauncher(); launcher.setSparkHome(System.getenv(SPARK_HOME_ENV)); + if (isCluster) { + launcher.addFile(fairSchedulerXMLPath); + } launcher.setAppResource(SparkLauncher.NO_RESOURCE); launcher.setPropertiesFile(confFile.getAbsolutePath()); launcher.setMainClass(RSCDriverBootstrapper.class.getName()); diff --git a/rsc/src/main/java/org/apache/livy/rsc/RSCConf.java b/rsc/src/main/java/org/apache/livy/rsc/RSCConf.java index 7c76164c0..b0e354b81 100644 --- a/rsc/src/main/java/org/apache/livy/rsc/RSCConf.java +++ b/rsc/src/main/java/org/apache/livy/rsc/RSCConf.java @@ -80,6 +80,8 @@ public static enum Entry implements ConfEntry { RETAINED_STATEMENTS("retained-statements", 100), RETAINED_SHARE_VARIABLES("retained.share-variables", 100), + CONCURRENT_SQL_SCHEDULER_POOL("concurrentSQL.scheduler.pool", "fair"), + CONCURRENT_SQL_MAX("concurrentSQL.max", 10), // Number of result rows to get for SQL Interpreters. SQL_NUM_ROWS("sql.num-rows", 1000);