Skip to content
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

(WIP) extend CREATE MODEL query to accept specifying config via file #17

Open
wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

(WIP) extend CREATE MODEL query to accept specifying config via file #17

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
8 changes: 7 additions & 1 deletion gateway/build.sbt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,14 @@ libraryDependencies ++= Seq(
"net.databinder.dispatch" %% "dispatch-core" % "0.11.2",
// parsing of program arguments
"com.github.scopt" %% "scopt" % "3.2.0",
// apache curator
"org.apache.curator" % "apache-curator" % "2.8.0",
"org.apache.curator" % "curator-framework" % "2.8.0",
"org.apache.curator" % "curator-recipes" % "2.8.0",
// testing
"org.scalatest" %% "scalatest" % "2.2.1"
"org.scalatest" %% "scalatest" % "2.2.1",
"org.apache.curator" % "curator-test" % "2.8.0",
"org.scala-lang.modules" %% "scala-async" % "0.9.2"
)

// disable parallel test execution to avoid BindException when mocking
Expand Down
9 changes: 7 additions & 2 deletions gateway/src/main/resources/log4j.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
<appender name="console" class="org.apache.log4j.ConsoleAppender">
<param name="Target" value="System.out"/>
<layout class="org.apache.log4j.PatternLayout">
<param name="ConversionPattern" value="%d{HH:mm:ss.SSS} [%t] %-5p %c{2} - %m%n"/>
<param name="ConversionPattern" value="%d{yyyy/MM/dd HH:mm:ss.SSS} [%t] %-5p %c{2} - %m%n"/>
</layout>
</appender>

Expand All @@ -16,7 +16,12 @@
<logger name="com.ning">
<level value="info"/>
</logger>

<logger name="org.apache.zookeeper">
<level value="error"/>
</logger>
<logger name="org.apache.curator">
<level value="error"/>
</logger>
<root>
<priority value="debug"/>
<appender-ref ref="console"/>
Expand Down
609 changes: 424 additions & 185 deletions gateway/src/main/scala/us/jubat/jubaql_server/gateway/GatewayPlan.scala
View file
Open in desktop

Large diffs are not rendered by default.

95 changes: 82 additions & 13 deletions gateway/src/main/scala/us/jubat/jubaql_server/gateway/JubaQLGateway.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,38 +20,66 @@ import scopt.OptionParser

object JubaQLGateway extends LazyLogging {
val defaultPort = 9877
val defaultThreads = 16
val defaultChannelMemory: Long = 65536
val defalutTotalMemory: Long = 1048576

/** Main function to start the JubaQL gateway application.
*/
def main(args: Array[String]) {
val maybeParsedOptions: Option[CommandlineOptions] = parseCommandlineOption(args)
if (maybeParsedOptions.isEmpty)
if (maybeParsedOptions.isEmpty) {
System.exit(1)
}
val parsedOptions = maybeParsedOptions.get

val ipAddress: String = parsedOptions.ip
val port: Int = parsedOptions.port

val gatewayId = parsedOptions.gatewayId match {
case "" => s"${ipAddress}_${port}"
case id => id
}

val persist: Boolean = parsedOptions.persist

var envp: Array[String] = Array()
var runMode: RunMode = RunMode.Development()
val runModeProperty: String = System.getProperty("run.mode")
val sparkJar = Option(System.getProperty("spark.yarn.jar"))
val zookeeperString = scala.util.Properties.propOrElse("jubaql.zookeeper", "")
val devModeRe = "development:([0-9]+)".r
val prodModeRe = "production:([0-9]+):([0-9]+)".r
val sparkDriverMemory = Option(System.getProperty("jubaql.processor.driverMemory"))
val sparkExecutorMemory = Option(System.getProperty("jubaql.processor.executorMemory"))

val devModeRe = "(local|development):([0-9]+)".r
val prodModeRe = "(cluster|production):([0-9]+):([0-9]+)".r
runModeProperty match {
case null | "" | "development" =>
case null | "" | "local" =>
runMode = RunMode.Development()

case "development" =>
logger.warn("run.mode use the 'local' rather than 'development'")
runMode = RunMode.Development()

case devModeRe(numThreadsString) =>
case devModeRe(modeString, numThreadsString) =>
if (modeString == "development") {
logger.warn("run.mode use the 'local' rather than 'development'")
}
runMode = RunMode.Development(numThreadsString.toInt)

case "production" =>
runMode = RunMode.Production(zookeeperString, sparkJar = sparkJar)
case "production" | "cluster" =>
if (runModeProperty.startsWith("production")) {
logger.warn("run.mode use the 'cluster' rather than 'production'")
}
runMode = RunMode.Production(zookeeperString, sparkJar = sparkJar,
sparkDriverMemory = sparkDriverMemory, sparkExecutorMemory = sparkExecutorMemory)

case prodModeRe(numExecutorsString, coresPerExecutorString) =>
case prodModeRe(modeString, numExecutorsString, coresPerExecutorString) =>
if (modeString == "production") {
logger.warn("run.mode use the 'cluster' rather than 'production'")
}
runMode = RunMode.Production(zookeeperString, numExecutorsString.toInt,
coresPerExecutorString.toInt, sparkJar = sparkJar)
coresPerExecutorString.toInt, sparkJar = sparkJar, sparkDriverMemory = sparkDriverMemory, sparkExecutorMemory = sparkExecutorMemory)

case _ =>
System.err.println("Bad run.mode property")
Expand All @@ -75,8 +103,17 @@ object JubaQLGateway extends LazyLogging {
"in production mode (comma-separated host:port list)")
System.exit(1)
}
case p: RunMode.Development =>
// When persist was configured, Set system property jubaql.zookeeper.
if (persist && zookeeperString.trim.isEmpty) {
logger.error("system property jubaql.zookeeper must be given " +
"with set persist flag (--persist)")
System.exit(1)
} else if (!persist && !zookeeperString.trim.isEmpty) {
logger.warn("persist flag is not specified; jubaql.zookeeper is ignored")
}
case _ =>
// don't set environment in dev mode
// don't set environment in other mode
}
logger.info("Starting in run mode %s".format(runMode))

Expand All @@ -86,10 +123,12 @@ object JubaQLGateway extends LazyLogging {
val plan = new GatewayPlan(ipAddress, port, envp, runMode,
sparkDistribution = sparkDistribution,
fatjar = fatjar,
checkpointDir = checkpointDir)
checkpointDir = checkpointDir, gatewayId, persist,
parsedOptions.threads, parsedOptions.channelMemory, parsedOptions.totalMemory)
val nettyServer = unfiltered.netty.Server.http(port).plan(plan)
logger.info("JubaQLGateway starting")
nettyServer.run()
plan.close()
logger.info("JubaQLGateway shut down successfully")
}

Expand All @@ -106,6 +145,35 @@ object JubaQLGateway extends LazyLogging {
x =>
if (x >= 1 && x <= 65535) success else failure("bad port number; port number n must be \"1 <= n <= 65535\"")
} text (f"port (default: $defaultPort%d)")
opt[String]('g', "gatewayID") optional() valueName ("<gatewayID>") action {
(x, o) =>
o.copy(gatewayId = x)
} text ("Gateway ID (default: ip_port)")
opt[Unit]("persist") optional() valueName ("<persist>") action {
(x, o) =>
o.copy(persist = true)
} text ("session persist")
opt[Int]("threads") optional() valueName ("<threads>") action {
(x, o) =>
o.copy(threads = x)
} validate {
x =>
if (x >= 1 && x <= Int.MaxValue) success else failure(s"invalid threads: specified in 1 or more and ${Int.MaxValue} or less")
} text (s"threads (default: $defaultThreads)")
opt[Long]("channel_memory") optional() valueName ("<channelMemory>") action {
(x, o) =>
o.copy(channelMemory = x)
} validate {
x =>
if (x >= 0 && x <= Long.MaxValue) success else failure(s"invalid channelMemory: specified in 0 or more and ${Long.MaxValue} or less")
} text (s"channelMemory (default: $defaultChannelMemory)")
opt[Long]("total_memory") optional() valueName ("<totalMemory>") action {
(x, o) =>
o.copy(totalMemory = x)
} validate {
x =>
if (x >= 0 && x <= Long.MaxValue) success else failure(s"invalid totalMemory: specified in 0 or more and ${Long.MaxValue} or less")
} text (s"totalMemory (default: $defalutTotalMemory)")
}

parser.parse(args, CommandlineOptions())
Expand All @@ -124,7 +192,7 @@ object JubaQLGateway extends LazyLogging {
val dir = scala.util.Properties.propOrElse("jubaql.checkpointdir", "")
if (dir.trim.isEmpty) {
runMode match {
case RunMode.Production(_, _, _, _) =>
case RunMode.Production(_, _, _, _, _, _) =>
"hdfs:///tmp/spark"
case RunMode.Development(_) =>
"file:///tmp/spark"
Expand All @@ -135,4 +203,5 @@ object JubaQLGateway extends LazyLogging {
}
}

case class CommandlineOptions(ip: String = "", port: Int = JubaQLGateway.defaultPort)
case class CommandlineOptions(ip: String = "", port: Int = JubaQLGateway.defaultPort, gatewayId: String = "", persist: Boolean = false,
threads: Int = JubaQLGateway.defaultThreads, channelMemory: Long = JubaQLGateway.defaultChannelMemory, totalMemory: Long = JubaQLGateway.defalutTotalMemory)
Loading