Skip to content

[LIVY-544] Allow interpreterExecutor run in ThreadPool #135

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
WeiWenda wants to merge 6 commits into
base: master
Choose a base branch
from
Open

[LIVY-544] Allow interpreterExecutor run in ThreadPool #135

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
01a9e0a
Allow interpreterExecutor run in ThreadPool
WeiWenda Dec 21, 2018
76d407b
Only SQL Kind Statement run in ThreadPool
WeiWenda Jan 15, 2019
3187b4e
Merge branch 'master' into livy-544
WeiWenda Feb 23, 2019
d384369
keep 'sql' serial execution, create 'concurrentSQL' to parallel execu…
WeiWenda Mar 12, 2019
97aa7c3
Merge branch 'master' into livy-544
WeiWenda Mar 13, 2019
936f554
add livy-fairscheduler.xml based on client/cluster
WeiWenda Mar 13, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 31 additions & 0 deletions conf/livy-fairscheduler.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
<?xml version="1.0"?>

<!--
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.
-->

<allocations>
<pool name="default">
<schedulingMode>FIFO</schedulingMode>
<weight>1</weight>
<minShare>0</minShare>
</pool>
<pool name="fair">
<schedulingMode>FAIR</schedulingMode>
<weight>1</weight>
<minShare>0</minShare>
</pool>
</allocations>
3 changes: 3 additions & 0 deletions core/src/main/scala/org/apache/livy/sessions/Kind.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@ object Shared extends Kind("shared")

object SQL extends Kind("sql")

object ConcurrentSQL extends Kind("concurrentSQL")

@mgaido91 mgaido91 Mar 15, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think that it is a good idea to add a new session kind. I think we should rather have a configuration for enabling FAIR vs FIFO and I am wondering if this shouldn't be done cross kind, ie. for all the kinds: why was it done only for SQL?

@Tagar Tagar Mar 15, 2019

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@mgaido91 good point. There was some discussion on this at https://issues.apache.org/jira/browse/LIVY-544
For code paragraphs chances are that there is a dependency between code items, and they can't run in parallel. For SQL sometimes there is no dependency often when it's just a SELECT.. but we let end-users decide if there is truly no dependency and let queries execute in parallel only when it was requested explicitly. Also this was modeled after Zeppelin to some degree - Zeppelin only allows parallel execution for SQL cells and not code cells like pyspark.

@WeiWenda WeiWenda Mar 15, 2019

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Run code block in parallel is very confusing. Spark‘s FAIR scheduler mode meaning that all runable stages can receive executors right away no matter when it was submitted. So that concurrentSQL’s stages need run into FAIR pool. But for scala or python or java, code block can't make stages in most time, it make no sense to let them run in FAIR pool.

@WeiWenda WeiWenda Mar 16, 2019

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I review 'SparkSqlInterperter' in zeppelin, it do the same thing as I have done.

NewSparkInterpreter

if (entry.getKey().toString().equals("zeppelin.spark.concurrentSQL")
            && entry.getValue().toString().equals("true")) {
          conf.set("spark.scheduler.mode", "FAIR");
        }

SparkSqlInterpreter

public InterpreterResult interpret(String st, InterpreterContext context)
      throws InterpreterException {
   ...
    sc.setLocalProperty("spark.scheduler.pool", context.getLocalProperties().get("pool"));

InterpreterContext's localProperties come from RemoteInterpreterContext's localProperties, which is RemoteInterpreter's InterpreterContext, created by Paragrah's getInterpreterContext, configured by %spark.sql(pool=poolname,key=value)

  private static Pattern REPL_PATTERN =
      Pattern.compile("(\\s*)%([\\w\\.]+)(\\(.*?\\))?.*", Pattern.DOTALL);

@WeiWenda WeiWenda Mar 16, 2019 •
edited
Loading

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But different from zeppelin, I don't think user can provide spark's fairscheduler.xml in proper way especially in cluster deploy, and can keep in mind which sql should run into which pool. This need a great understand of spark scheduler. So I do all things in livy to release user from configuring above items. Of course we can modify to

{"code":"", "kind":"sql", "pool":"xx"}
//or
{"code":"set pool=xx, query content", "kind":"sql"}
//or
{"code":"", "kind":"sql?pool=xx"}

I have to say that expansibility of livy is not as excellent as zeppelin, there should be a properties part in statement boby.
image


object Kind {

def apply(kind: String): Kind = kind match {
Expand All @@ -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")
}
}
Expand Down
5 changes: 4 additions & 1 deletion repl/src/main/scala/org/apache/livy/repl/SQLInterpreter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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)

Expand All @@ -94,6 +96,7 @@ class SQLInterpreter(
case e => e
}
}
spark.sparkContext.setLocalProperty("spark.scheduler.pool", null)

val jRows = Extraction.decompose(rows)

Expand Down
17 changes: 14 additions & 3 deletions repl/src/main/scala/org/apache/livy/repl/Session.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())

Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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) {
Expand All @@ -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
}
Expand Down Expand Up @@ -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")""",
Expand Down
19 changes: 19 additions & 0 deletions rsc/src/main/java/org/apache/livy/rsc/ContextLauncher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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());
Expand Down
2 changes: 2 additions & 0 deletions rsc/src/main/java/org/apache/livy/rsc/RSCConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand Down