Блохин Леонид - "mist, как часть hydrosphere"

Misthttps://github.com/Hydrospheredata/mist

www.provectus.com

© Provectus, Inc.

1

• Леонид Блохин

• Big Data Engineer

• lblokhin@provectus.com

• +7 (917) 295 - 40 - 49

Mist

• HydroSphere

• Spark

• Why We Needed a Mist

• Running

• Configuration

• Spark Job at Mist

• Road Map

www.provectus.com2

Mist

www.provectus.com3

http://hydrosphere.io/

Hydrosphere – Opensource Big Data and Analytics platform

with DevOps culture in mind.

Mist

www.provectus.com4

http://hydrosphere.io/

Mist

www.provectus.com5

http://spark.apache.org/

Apache Spark™ is a fast and general engine for large-scale data processing.

Mist

www.provectus.com6

Mist

• Mist is a thin service on top of Spark which makes it possible to execute Scala & Python Spark Jobs

from application layers and get synchronous, asynchronous, and reactive results as well as provide

an API to external clients.

• It implements Spark as a Service and creates a unified API layer for building enterprise solutions

and services on top of a Big Data lake.

www.provectus.com7

Mist

● HTTP and Messaging (MQTT) API● Scala & Python Spark job execution● Works with Standalone, Mesos, Yarn any Spark config● Support for Spark SQL and Hive● High Availability and Fault Tolerance● Persist job state for self healing● Async and sync API, JSON job results

www.provectus.com8

Why We Needed a Mist

Mist

Build the project

git clone https://github.com/hydrospheredata/mist.git

cd mist

./sbt/sbt -DsparkVersion=1.5.2 assembly

Create configuration file

Run

spark-submit --class io.hydrosphere.mist.Mist \

--driver-java-options "-Dconfig.file=/path/to/application.conf" \

target/scala-2.10/mist-assembly-0.2.0.jar

www.provectus.com9

Running

Mist

www.provectus.com

Configuration

10

# spark master url can be either of three: local, yarn, mesos (local by default)mist.spark.master = "local[*]"

# number of threads: one thread for one jobmist.settings.threadNumber = 16

# http interface (off by default)mist.http.on = truemist.http.host = "192.168.10.13"mist.http.port = 2003

Mist

www.provectus.com

Configuration

11

# MQTT interface (off by default)mist.mqtt.on = truemist.mqtt.host = "192.168.10.33"mist.mqtt.port = 1883# mist listens this topic for incoming requestsmist.mqtt.subscribeTopic = "foo"# mist answers in this topic with the resultsmist.mqtt.publishTopic = "foo"

Mist

www.provectus.com

Configuration

12

# recovery job (off by default)mist.recovery.on = truemist.recovery.multilimit = 10mist.recovery.typedb = "MapDb"mist.recovery.dbfilename = "file.db"

Mist

www.provectus.com

Configuration

13

# default settings for all contexts# timeout for each job in contextmist.contextDefaults.timeout = 100 days# mist can kill context after job finished (off by default)mist.contextDefaults.disposable = false

# settings for SparkConfmist.contextDefaults.sparkConf = { spark.default.parallelism = 128 spark.driver.memory = "10g" spark.scheduler.mode = "FAIR"}

Mist

www.provectus.com

Configuration

14

# settings can be overridden for each contextmist.contexts.foo.timeout = 100 days

mist.contexts.foo.sparkConf = { spark.scheduler.mode = "FIFO"}

mist.contexts.bar.timeout = 1000 secondmist.contexts.bar.disposable = true

# mist can create context on start, so we don't waste time on first requestmist.contextSettings.onstart = ["foo"]

Mist

Spark Job at MistMist Scala Spark Job

In order to prepare your job to run on Mist you should extend scala object from MistJob and implement abstract method

doStuff :

def doStuff(context: SparkContext, parameters: Map[String, Any]): Map[String, Any] = ???

def doStuff(context: SQLContext, parameters: Map[String, Any]): Map[String, Any] = ???

def doStuff(context: HiveContext, parameters: Map[String, Any]): Map[String, Any] = ???

www.provectus.com15

Mist

Spark Job at MistExample:

object SimpleContext extends MistJob {

override def doStuff(context: SparkContext, parameters: Map[String, Any]): Map[String, Any] = {

val numbers: List[BigInt] = parameters("digits").asInstanceOf[List[BigInt]]

val rdd = context.parallelize(numbers)

Map("result" -> rdd.map(x => x * 2).collect())

}

}

Building Mist jobs

Add Mist as dependency in your build.sbt:libraryDependencies += "io.hydrosphere" % "mist" % "0.2.0"

www.provectus.com16

Mist

Spark Job at MistMist Python Spark Job

Import mist and implement method doStuff.The following are Spark Contexts aliases to be used for convenience:

job.sc = SparkContext

job.sqlc = SQL Context

job.hc = Hive Context

www.provectus.com17

Mist

Spark Job at Mist

for examplimport mist

class MyJob:

def __init__(self, job):

job.sendResult(self.doStuff(job))

def doStuff(self, job):

val = job.parameters.values()

list = val.head()

pylist = []

count = 0

while count < list.size():

pylist.append(list.head())

count = count + 1

list = list.tail()

rdd = job.sc.parallelize(pylist)

result = rdd.map(lambda s: 2 * s).collect()

return result

if __name__ == "__main__":

job = MyJob(mist.Job())

www.provectus.com18

Mist

www.provectus.com19

mosquitto_pub -h 192.168.10.33 -p 1883 -m'{

"jarPath":"/vagrant/examples/target/scala-2.11/mist_examples_2.11-0.0.1.jar", "className":"SimpleContext$","parameters":{"digits":[1,2,3,4,5,6,7,8,9,0]}, "external_id":"12345678","name":"foo"

}' -t 'foo'

Mist

www.provectus.com20

Mist

www.provectus.com21

{"success":true,"payload":{"result":[2,4,6,8,10,12,14,16,18,0]},"errors":[],"request":{"jarPath":"src/test/resources/mistjob_2.10-1.0.jar","className":"SimpleContext$","name":"foo","parameters":{"digits":[1,2,3,4,5,6,7,8,9,0]},"external_id":"12345678"}

}

Mist

www.provectus.com22

● Super parallel mode Support multi JVM● Cluster mode and node framework● Add logging● Restification● Support streaming contexts/jobs● Apache Kafka support● AMQP support● Web UI

Your contributions are very welcome on Github!https://github.com/Hydrospheredata/mist

Road Map

Thanks!

Questions?

www.provectus.com23

Леонид Блохин

Skype: leonid_niko

Email: lblokhin@provectus.com

www.provectus.com