From 01a9e0ab5dfd53c5337c0e52247fd9682795bc75 Mon Sep 17 00:00:00 2001 From: weiwenda Date: Fri, 21 Dec 2018 11:24:21 +0800 Subject: [PATCH 1/4] Allow interpreterExecutor run in ThreadPool --- repl/src/main/scala/org/apache/livy/repl/Session.scala | 4 ++-- rsc/src/main/java/org/apache/livy/rsc/RSCConf.java | 1 + 2 files changed, 3 insertions(+), 2 deletions(-) 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..78df50041 100644 --- a/repl/src/main/scala/org/apache/livy/repl/Session.scala +++ b/repl/src/main/scala/org/apache/livy/repl/Session.scala @@ -58,7 +58,7 @@ class Session( import Session._ private val interpreterExecutor = ExecutionContext.fromExecutorService( - Executors.newSingleThreadExecutor()) + Executors.newFixedThreadPool(livyConf.getInt(RSCConf.Entry.SESSION_INTERPRETER_THREADS))) private val cancelExecutor = ExecutionContext.fromExecutorService( Executors.newSingleThreadExecutor()) @@ -161,7 +161,7 @@ class Session( _statements.synchronized { _statements(statementId) = statement } Future { - setJobGroup(tpe, statementId) + this.synchronized { setJobGroup(tpe, statementId) } statement.compareAndTransit(StatementState.Waiting, StatementState.Running) if (statement.state.get() == StatementState.Running) { 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..fd645fd22 100644 --- a/rsc/src/main/java/org/apache/livy/rsc/RSCConf.java +++ b/rsc/src/main/java/org/apache/livy/rsc/RSCConf.java @@ -44,6 +44,7 @@ public static enum Entry implements ConfEntry { CLIENT_SHUTDOWN_TIMEOUT("client.shutdown-timeout", "10s"), DRIVER_CLASS("driver-class", null), SESSION_KIND("session.kind", null), + SESSION_INTERPRETER_THREADS("session.interpreter.threadPool.size", 1), LIVY_JARS("jars", null), SPARKR_PACKAGE("sparkr.package", null), From 76d407b356773ec5b5e3d2cca77f0eabfeca8bd4 Mon Sep 17 00:00:00 2001 From: weiwenda Date: Tue, 15 Jan 2019 10:56:56 +0800 Subject: [PATCH 2/4] Only SQL Kind Statement run in ThreadPool --- bin/livy-server | 2 +- conf/livy-fairscheduler.xml | 31 +++++++++++++++++++ .../scala/org/apache/livy/sessions/Kind.scala | 3 ++ .../org/apache/livy/repl/SQLInterpreter.scala | 5 ++- .../scala/org/apache/livy/repl/Session.scala | 19 +++++++++--- .../org/apache/livy/rsc/ContextLauncher.java | 6 ++++ .../java/org/apache/livy/rsc/RSCConf.java | 3 +- 7 files changed, 62 insertions(+), 7 deletions(-) create mode 100644 conf/livy-fairscheduler.xml diff --git a/bin/livy-server b/bin/livy-server index 8d27d4e61..70643eb8d 100755 --- a/bin/livy-server +++ b/bin/livy-server @@ -21,7 +21,7 @@ usage="Usage: livy-server (start|stop|status)" export LIVY_HOME=$(cd $(dirname $0)/.. && pwd) -LIVY_CONF_DIR=${LIVY_CONF_DIR:-"$LIVY_HOME/conf"} +export LIVY_CONF_DIR=${LIVY_CONF_DIR:-"$LIVY_HOME/conf"} if [ -f "${LIVY_CONF_DIR}/livy-env.sh" ]; then # Promote all variable declarations to environment (exported) variables 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..056318024 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 SerialSQL extends Kind("serial_sql") + 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 "serial_sql" => SerialSQL 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 78df50041..3a833077d 100644 --- a/repl/src/main/scala/org/apache/livy/repl/Session.scala +++ b/repl/src/main/scala/org/apache/livy/repl/Session.scala @@ -58,7 +58,10 @@ class Session( import Session._ private val interpreterExecutor = ExecutionContext.fromExecutorService( - Executors.newFixedThreadPool(livyConf.getInt(RSCConf.Entry.SESSION_INTERPRETER_THREADS))) + Executors.newSingleThreadExecutor()) + + private lazy val sqlExecutor = ExecutionContext.fromExecutorService( + Executors.newFixedThreadPool(livyConf.getInt(RSCConf.Entry.SQL_INTERPRETER_THREADS))) private val cancelExecutor = ExecutionContext.fromExecutorService( Executors.newSingleThreadExecutor()) @@ -105,7 +108,9 @@ class Session( throw new IllegalStateException("SparkInterpreter should not be lazily created.") case PySpark => PythonInterpreter(sparkConf, entries) case SparkR => SparkRInterpreter(sparkConf, entries) - case SQL => new SQLInterpreter(sparkConf, livyConf, entries) + case SQL => new SQLInterpreter(sparkConf, livyConf, entries, + livyConf.get(RSCConf.Entry.SQL_DEFAULT_SCHEDULER_POOL)) + case SerialSQL => new SQLInterpreter(sparkConf, livyConf, entries) } interp.start() interpGroup(kind) = interp @@ -160,6 +165,12 @@ class Session( val statement = new Statement(statementId, code, StatementState.Waiting, null) _statements.synchronized { _statements(statementId) = statement } + val threadPool = if (tpe == SQL) { + sqlExecutor + } else { + interpreterExecutor + } + Future { this.synchronized { setJobGroup(tpe, statementId) } statement.compareAndTransit(StatementState.Waiting, 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 | SerialSQL => // 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 790f912cf..e220f8b74 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 { @@ -180,6 +181,9 @@ 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"); + conf.set("spark.scheduler.allocation.file", FAIR_SCHEDULER_FILENAME); + // 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"); @@ -219,6 +223,8 @@ public void run() { } else { final SparkLauncher launcher = new SparkLauncher(); launcher.setSparkHome(System.getenv(SPARK_HOME_ENV)); + launcher.addFile( + System.getenv().get("LIVY_CONF_DIR") + File.separator + FAIR_SCHEDULER_FILENAME); 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 fd645fd22..b6204e52f 100644 --- a/rsc/src/main/java/org/apache/livy/rsc/RSCConf.java +++ b/rsc/src/main/java/org/apache/livy/rsc/RSCConf.java @@ -44,7 +44,6 @@ public static enum Entry implements ConfEntry { CLIENT_SHUTDOWN_TIMEOUT("client.shutdown-timeout", "10s"), DRIVER_CLASS("driver-class", null), SESSION_KIND("session.kind", null), - SESSION_INTERPRETER_THREADS("session.interpreter.threadPool.size", 1), LIVY_JARS("jars", null), SPARKR_PACKAGE("sparkr.package", null), @@ -81,6 +80,8 @@ public static enum Entry implements ConfEntry { RETAINED_STATEMENTS("retained-statements", 100), RETAINED_SHARE_VARIABLES("retained.share-variables", 100), + SQL_DEFAULT_SCHEDULER_POOL("sql.default.scheduler.pool", "fair"), + SQL_INTERPRETER_THREADS("sql.interpreter.threadPool.size", 1), // Number of result rows to get for SQL Interpreters. SQL_NUM_ROWS("sql.num-rows", 1000); From d3843691f0cd1ec2008ef86622f25a90bd3cd3bb Mon Sep 17 00:00:00 2001 From: weiwenda Date: Tue, 12 Mar 2019 12:17:23 +0800 Subject: [PATCH 3/4] keep 'sql' serial execution, create 'concurrentSQL' to parallel execution --- .../scala/org/apache/livy/sessions/Kind.scala | 4 ++-- .../scala/org/apache/livy/repl/Session.scala | 16 ++++++++-------- .../main/java/org/apache/livy/rsc/RSCConf.java | 4 ++-- 3 files changed, 12 insertions(+), 12 deletions(-) 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 056318024..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,7 +35,7 @@ object Shared extends Kind("shared") object SQL extends Kind("sql") -object SerialSQL extends Kind("serial_sql") +object ConcurrentSQL extends Kind("concurrentSQL") object Kind { @@ -45,7 +45,7 @@ object Kind { case "sparkr" | "r" => SparkR case "shared" => Shared case "sql" => SQL - case "serial_sql" => SerialSQL + case "concurrentSQL" => ConcurrentSQL case other => throw new IllegalArgumentException(s"Invalid kind: $other") } } 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 3a833077d..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,8 +60,8 @@ class Session( private val interpreterExecutor = ExecutionContext.fromExecutorService( Executors.newSingleThreadExecutor()) - private lazy val sqlExecutor = ExecutionContext.fromExecutorService( - Executors.newFixedThreadPool(livyConf.getInt(RSCConf.Entry.SQL_INTERPRETER_THREADS))) + private lazy val concurrentSQLExecutor = ExecutionContext.fromExecutorService( + Executors.newFixedThreadPool(livyConf.getInt(RSCConf.Entry.CONCURRENT_SQL_MAX))) private val cancelExecutor = ExecutionContext.fromExecutorService( Executors.newSingleThreadExecutor()) @@ -108,9 +108,9 @@ class Session( throw new IllegalStateException("SparkInterpreter should not be lazily created.") case PySpark => PythonInterpreter(sparkConf, entries) case SparkR => SparkRInterpreter(sparkConf, entries) - case SQL => new SQLInterpreter(sparkConf, livyConf, entries, - livyConf.get(RSCConf.Entry.SQL_DEFAULT_SCHEDULER_POOL)) - case SerialSQL => new SQLInterpreter(sparkConf, livyConf, 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() interpGroup(kind) = interp @@ -165,8 +165,8 @@ class Session( val statement = new Statement(statementId, code, StatementState.Waiting, null) _statements.synchronized { _statements(statementId) = statement } - val threadPool = if (tpe == SQL) { - sqlExecutor + val threadPool = if (tpe == ConcurrentSQL) { + concurrentSQLExecutor } else { interpreterExecutor } @@ -344,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 | SerialSQL => + 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/RSCConf.java b/rsc/src/main/java/org/apache/livy/rsc/RSCConf.java index b6204e52f..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,8 +80,8 @@ public static enum Entry implements ConfEntry { RETAINED_STATEMENTS("retained-statements", 100), RETAINED_SHARE_VARIABLES("retained.share-variables", 100), - SQL_DEFAULT_SCHEDULER_POOL("sql.default.scheduler.pool", "fair"), - SQL_INTERPRETER_THREADS("sql.interpreter.threadPool.size", 1), + 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); From 936f55415728afeca3263318402559f5cc862ea7 Mon Sep 17 00:00:00 2001 From: weiwenda <994184916@qq.com> Date: Wed, 13 Mar 2019 23:48:41 +0800 Subject: [PATCH 4/4] add livy-fairscheduler.xml based on client/cluster --- bin/livy-server | 2 +- .../org/apache/livy/rsc/ContextLauncher.java | 19 ++++++++++++++++--- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/bin/livy-server b/bin/livy-server index 70643eb8d..8d27d4e61 100755 --- a/bin/livy-server +++ b/bin/livy-server @@ -21,7 +21,7 @@ usage="Usage: livy-server (start|stop|status)" export LIVY_HOME=$(cd $(dirname $0)/.. && pwd) -export LIVY_CONF_DIR=${LIVY_CONF_DIR:-"$LIVY_HOME/conf"} +LIVY_CONF_DIR=${LIVY_CONF_DIR:-"$LIVY_HOME/conf"} if [ -f "${LIVY_CONF_DIR}/livy-env.sh" ]; then # Promote all variable declarations to environment (exported) variables 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 c32bdb981..4e5616add 100644 --- a/rsc/src/main/java/org/apache/livy/rsc/ContextLauncher.java +++ b/rsc/src/main/java/org/apache/livy/rsc/ContextLauncher.java @@ -202,7 +202,19 @@ private static ChildProcess startDriver(final RSCConf conf, Promise promise) conf.set("spark.yarn.maxAppAttempts", "1"); conf.set("spark.scheduler.mode", "FAIR"); - conf.set("spark.scheduler.allocation.file", FAIR_SCHEDULER_FILENAME); + 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. @@ -243,8 +255,9 @@ public void run() { } else { final SparkLauncher launcher = new SparkLauncher(); launcher.setSparkHome(System.getenv(SPARK_HOME_ENV)); - launcher.addFile( - System.getenv().get("LIVY_CONF_DIR") + File.separator + FAIR_SCHEDULER_FILENAME); + if (isCluster) { + launcher.addFile(fairSchedulerXMLPath); + } launcher.setAppResource(SparkLauncher.NO_RESOURCE); launcher.setPropertiesFile(confFile.getAbsolutePath()); launcher.setMainClass(RSCDriverBootstrapper.class.getName());