Welcome to Livy, the REST Spark Server ====================================== Livy is an open source REST interface for interacting with Spark from anywhere. It supports executing snippets of code or programs in a Spark context that runs locally or in YARN. * Interactive Scala, Python and R shells * Batch submissions in Scala, Java, Python * Multi users can share the same server (impersonation support) * Can be used for submitting jobs from anywhere with REST * Does not require any code change to your programs The code is currently incubating in Hue but hopefully will eventually graduate in its top project. `Pull requests` are welcomed! .. _Pull requests: https://github.com/cloudera/hue/pulls Quick Start ============= Livy is used for powering the Spark snippets of the `Hadoop Notebook`_ of `Hue 3.8`_, which you can see the `implementation here`_. See the API documentation below and some curl examples: * `Interactive shells`_ * `Batch jobs`_ * `Shared RDDs`_ .. _Interactive shells: http://gethue.com/how-to-use-the-livy-spark-rest-job-server-for-interactive-spark/ .. _Batch jobs: http://gethue.com/how-to-use-the-livy-spark-rest-job-server-api-for-sharing-spark-rdds-and-contexts/ .. _Shared RDDs: http://gethue.com/how-to-use-the-livy-spark-rest-job-server-api-for-submitting-batch-jar-python-and-streaming-spark-jobs/ .. _Hadoop Notebook: http://gethue.com/new-notebook-application-for-spark-sql/ .. _Hue 3.8: http://gethue.com/hue-3-8-with-an-oozie-editor-revamp-better-performances-improved-spark-ui-is-out/ .. _implementation here: https://github.com/cloudera/hue/blob/master/apps/spark/src/spark/job_server_api.py Prerequisites ============= To build/run Livy, you will need: Debian/Ubuntu: * mvn (from ``maven`` package or maven3 tarball) * openjdk-7-jdk (or Oracle Java7 jdk) * spark 1.5 from (from `Apache Spark tarball`_) * Python 2.6+ * R 3.x Redhat/CentOS: * mvn (from ``maven`` package or maven3 tarball) * java-1.7.0-openjdk (or Oracle Java7 jdk) * spark 1.5 (from `Apache Spark tarball`_) * Python 2.6+ * R 3.x MacOS: * Xcode command line tools * Oracle's JDK 1.7+ * Maven (Homebrew) * apache-spark 1.5 (Homebrew) * Python 2.6+ * R 3.x .. _Apache Spark Tarball: https://spark.apache.org/downloads.html Building Livy ============= Livy is currently built by the `Hue Build System`_, it can also be built on it's own (aka without any other Hue dependency) with `Apache Maven`_. To build, checks out the code, go to the Livy directory and run: .. code:: shell git clone git@github.com:cloudera/hue.git cd hue .. code:: shell % cd apps/spark/java % mvn -DskipTests clean package .. _Hue Build System: https://github.com/cloudera/hue/#getting-started .. _Apache Maven: http://maven.apache.org Running Tests ============= In order to run the Livy Tests, first follow the instructions in `Building Livy`_. Then run: .. code:: shell % export SPARK_HOME=/usr/lib/spark % export HADOOP_CONF_DIR=/etc/hadoop/conf % mvn test Running Livy ============ In order to run Livy with local sessions, first export these variables: .. code:: shell % export SPARK_HOME=/usr/lib/spark % export HADOOP_CONF_DIR=/etc/hadoop/conf Then start the server with: .. code:: shell % ./bin/livy-server Or with YARN sessions by running: .. code:: shell % env \ LIVY_SERVER_JAVA_OPTS="-Dlivy.server.session.factory=yarn" \ CLASSPATH=`hadoop classpath` \ ./bin/livy-server Livy Configuration ================== The properties of the server can be modified by copying and renaming it ``livy-defaults.conf``. In particular the ``YARN mode`` (default is ``local`` process for development) can be set with: .. code:: shell livy.server.session.factory = yarn Spark Example ============= Now to see it in action by interacting with it in Python with the `Requests`_ library. By default livy runs on port 8998 (which can be changed with the ``livy_server_port config`` option). We’ll start off with a Spark session that takes Scala code: .. code:: shell % sudo pip install requests .. code:: python >>> import json, pprint, requests, textwrap >>> host = 'http://localhost:8998' >>> data = {'kind': 'spark'} >>> headers = {'Content-Type': 'application/json'} >>> r = requests.post(host + '/sessions', data=json.dumps(data), headers=headers) >>> r.json() {u'state': u'starting', u'id': 0, u’kind’: u’spark’} Once the session has completed starting up, it transitions to the idle state: .. code:: python >>> session_url = host + r.headers['location'] >>> r = requests.get(session_url, headers=headers) >>> r.json() {u'state': u'idle', u'id': 0, u’kind’: u’spark’} Now we can execute Scala by passing in a simple JSON command: .. code:: python >>> statements_url = session_url + '/statements' >>> data = {'code': '1 + 1'} >>> r = requests.post(statements_url, data=json.dumps(data), headers=headers) >>> r.json() {u'output': None, u'state': u'running', u'id': 0} If a statement takes longer than a few milliseconds to execute, Livy returns early and provides a URL that can be polled until it is complete: .. code:: python >>> statement_url = host + r.headers['location'] >>> r = requests.get(statement_url, headers=headers) >>> pprint.pprint(r.json()) [{u'id': 0, u'output': {u'data': {u'text/plain': u'res0: Int = 2'}, u'execution_count': 0, u'status': u'ok'}, u'state': u'available'}] That was a pretty simple example. More interesting is using Spark to estimate PI. This is from the `Spark Examples`_: .. code:: python >>> data = { ... 'code': textwrap.dedent("""\ ... val NUM_SAMPLES = 100000; ... val count = sc.parallelize(1 to NUM_SAMPLES).map { i => ... val x = Math.random(); ... val y = Math.random(); ... if (x*x + y*y < 1) 1 else 0 ... }.reduce(_ + _); ... println(\"Pi is roughly \" + 4.0 * count / NUM_SAMPLES) ... """) ... } >>> r = requests.post(statements_url, data=json.dumps(data), headers=headers) >>> pprint.pprint(r.json()) {u'id': 1, u'output': {u'data': {u'text/plain': u'Pi is roughly 3.14004\nNUM_SAMPLES: Int = 100000\ncount: Int = 78501'}, u'execution_count': 1, u'status': u'ok'}, u'state': u'available'} Finally, lets close our session: .. code:: python >>> session_url = 'http://localhost:8998/sessions/0' >>> requests.delete(session_url, headers=headers) .. _Requests: http://docs.python-requests.org/en/latest/ .. _Spark Examples: https://spark.apache.org/examples.html PySpark Example =============== pyspark has the exact same API, just with a different initial command: .. code:: python >>> data = {'kind': 'pyspark'} >>> r = requests.post(host + '/sessions', data=json.dumps(data), headers=headers) >>> r.json() {u'id': 1, u'state': u'idle'} The PI example from before then can be run as: .. code:: python >>> data = { ... 'code': textwrap.dedent("""\ ... import random ... NUM_SAMPLES = 100000 ... def sample(p): ... x, y = random.random(), random.random() ... return 1 if x*x + y*y < 1 else 0 ... ... count = sc.parallelize(xrange(0, NUM_SAMPLES)).map(sample) \ ... .reduce(lambda a, b: a + b) ... print "Pi is roughly %f" % (4.0 * count / NUM_SAMPLES) ... """) ... } >>> r = requests.post(statements_url, data=json.dumps(data), headers=headers) >>> pprint.pprint(r.json()) {u'id': 12, u'output': {u'data': {u'text/plain': u'Pi is roughly 3.136000'}, u'execution_count': 12, u'status': u'ok'}, u'state': u'running'} SparkR Example ============== SparkR also has the same API: .. code:: python >>> data = {'kind': 'sparkR'} >>> r = requests.post(host + '/sessions', data=json.dumps(data), headers=headers) >>> r.json() {u'id': 1, u'state': u'idle'} The PI example from before then can be run as: .. code:: python >>> data = { ... 'code': textwrap.dedent("""\ ... n <- 100000 ... piFunc <- function(elem) { ... rands <- runif(n = 2, min = -1, max = 1) ... val <- ifelse((rands[1]^2 + rands[2]^2) < 1, 1.0, 0.0) ... val ... } ... piFuncVec <- function(elems) { ... message(length(elems)) ... rands1 <- runif(n = length(elems), min = -1, max = 1) ... rands2 <- runif(n = length(elems), min = -1, max = 1) ... val <- ifelse((rands1^2 + rands2^2) < 1, 1.0, 0.0) ... sum(val) ... } ... rdd <- parallelize(sc, 1:n, slices) ... count <- reduce(lapplyPartition(rdd, piFuncVec), sum) ... cat("Pi is roughly", 4.0 * count / n, "\n") ... """) ... } >>> r = requests.post(statements_url, data=json.dumps(data), headers=headers) >>> pprint.pprint(r.json()) {u'id': 12, u'output': {u'data': {u'text/plain': u'Pi is roughly 3.136000'}, u'execution_count': 12, u'status': u'ok'}, u'state': u'running'} Community ========= * User group: http://groups.google.com/a/cloudera.org/group/hue-user * Umbrella Jira: https://issues.cloudera.org/browse/HUE-2588 * Pull requests: https://github.com/cloudera/hue/pulls REST API ======== GET /sessions ------------- Returns all the active interactive sessions. Response Body ^^^^^^^^^^^^^ +----------+-----------------+------+ | name | description | type | +==========+=================+======+ | sessions | `session`_ list | list | +----------+-----------------+------+ POST /sessions -------------- Creates a new interative Scala, Python or R shell in the cluster. Request Body ^^^^^^^^^^^^ +----------------+--------------------------------------------------------------------------------+------------------+ | name | description | type | +================+================================================================================+==================+ | kind | The session kind (required) | `session kind`_ | +----------------+--------------------------------------------------------------------------------+------------------+ | proxyUser | The user to impersonate that will run this session (e.g. bob) | string | +----------------+--------------------------------------------------------------------------------+------------------+ | jars | Files to be placed on the java classpath | list of paths | +----------------+--------------------------------------------------------------------------------+------------------+ | pyFiles | Files to be placed on the PYTHONPATH | list of paths | +----------------+--------------------------------------------------------------------------------+------------------+ | files | Files to be placed in executor working directory | list of paths | +----------------+--------------------------------------------------------------------------------+------------------+ | driverMemory | Memory for driver (e.g. 1000M, 2G) | string | +----------------+--------------------------------------------------------------------------------+------------------+ | driverCores | Number of cores used by driver (YARN mode only) | int | +----------------+--------------------------------------------------------------------------------+------------------+ | executorMemory | Memory for executor (e.g. 1000M, 2G) | string | +----------------+--------------------------------------------------------------------------------+------------------+ | executorCores | Number of cores used by executor | int | +----------------+--------------------------------------------------------------------------------+------------------+ | numExecutors | Number of executors (YARN mode only) | int | +----------------+--------------------------------------------------------------------------------+------------------+ | archives | Archives to be uncompressed in the executor working directory (YARN mode only) | list of paths | +----------------+--------------------------------------------------------------------------------+------------------+ | queue | The YARN queue to submit too (YARN mode only) | string | +----------------+--------------------------------------------------------------------------------+------------------+ | name | Name of the application | string | +----------------+--------------------------------------------------------------------------------+------------------+ | conf | Spark configuration property | list of key=val | +----------------+--------------------------------------------------------------------------------+------------------+ Response Body ^^^^^^^^^^^^^ The created `Session`_. GET /sessions/{sessionId} ------------------------- Return the session information Response ^^^^^^^^ The `Session`_. DELETE /sessions/{sessionId} ------------------------- Kill the `Session`_ job. GET /sessions/{sessionId}/logs --------------------------- Get the log lines from this session. Request Parameters ^^^^^^^^^^^^^^^^^^ +------+-----------------------------+------+ | name | description | type | +======+=============================+======+ | from | offset | int | +------+-----------------------------+------+ | size | amount of batches to return | int | +------+-----------------------------+------+ Response Body ^^^^^^^^^^^^^ +------+-----------------------+-----------------+ | name | description | type | +======+=======================+=================+ | id | The session id | int | +------+-----------------------+-----------------+ | from | offset | int | +------+-----------------------+-----------------+ | size | total amount of lines | int | +------+-----------------------+-----------------+ | log | The log lines | list of strings | +------+-----------------------+-----------------+ GET /sessions/{sessionId}/statements ------------------------------------ Return all the statements in a session. Response Body ^^^^^^^^^^^^^ +------------+-------------------+------+ | name | description | type | +============+===================+======+ | statements | `statement`_ list | list | +------------+-------------------+------+ POST /sessions/{sessionId}/statements ------------------------------------- Execute a statement in a session. Request Body ^^^^^^^^^^^^ +------+---------------------+--------+ | name | description | type | +======+=====================+========+ | code | The code to execute | string | +------+---------------------+--------+ Response Body ^^^^^^^^^^^^^ The `statement`_ object. GET /batches ------------ Return all the active batch jobs. Response Body ^^^^^^^^^^^^^ +---------+---------------+------+ | name | description | type | +=========+===============+======+ | batches | `batch`_ list | list | +---------+---------------+------+ POST /batches ------------- Request Body ^^^^^^^^^^^^ +----------------+--------------------------------------------------+-----------------+ | name | description | type | +================+==================================================+=================+ | proxyUser | The user to impersonate that will execute the job| string | +----------------+--------------------------------------------------+-----------------+ | file | Archive holding the file | path (required) | +----------------+--------------------------------------------------+-----------------+ | args | Command line arguments | list of strings | +----------------+--------------------------------------------------+-----------------+ | className | Application's java/spark main class | string | +----------------+--------------------------------------------------+-----------------+ | jars | Files to be placed on the java classpath | list of paths | +----------------+--------------------------------------------------+-----------------+ | pyFiles | Files to be placed on the PYTHONPATH | list of paths | +----------------+--------------------------------------------------+-----------------+ | files | Files to be placed in executor working directory | list of paths | +----------------+--------------------------------------------------+-----------------+ | driverMemory | Memory for driver (e.g. 1000M, 2G) | string | +----------------+--------------------------------------------------+-----------------+ | driverCores | Number of cores used by driver | int | +----------------+--------------------------------------------------+-----------------+ | executorMemory | Memory for executor (e.g. 1000M, 2G) | string | +----------------+--------------------------------------------------+-----------------+ | executorCores | Number of cores used by executor | int | +----------------+--------------------------------------------------+-----------------+ | numExecutors | Number of executor | int | +----------------+--------------------------------------------------+-----------------+ | archives | Archives to be uncompressed (YARN mode only) | list of paths | +----------------+--------------------------------------------------+-----------------+ | queue | The YARN queue to submit too (YARN mode only) | string | +----------------+--------------------------------------------------+-----------------+ | name | Name of the application | string | +----------------+--------------------------------------------------+-----------------+ | conf | Spark configuration property | list of key=val | +----------------+--------------------------------------------------+-----------------+ Response Body ^^^^^^^^^^^^^ The created `Batch`_ object. GET /batches/{batchId} ---------------------- Request Parameters ^^^^^^^^^^^^^^^^^^ +------+-----------------------------+------+ | name | description | type | +======+=============================+======+ | from | offset | int | +------+-----------------------------+------+ | size | amount of batches to return | int | +------+-----------------------------+------+ Response Body ^^^^^^^^^^^^^ +-------+-----------------------------+-----------------+ | name | description | type | +=======+=============================+=================+ | id | The batch id | int | +-------+-----------------------------+-----------------+ | state | The state of the batch | `batch`_ state | +-------+-----------------------------+-----------------+ | log | The output of the batch job | list of strings | +-------+-----------------------------+-----------------+ DELETE /batches/{batchId} ------------------------- Kill the `Batch`_ job. GET /batches/{batchId}/logs --------------------------- Get the log lines from this batch. Request Parameters ^^^^^^^^^^^^^^^^^^ +------+-----------------------------+------+ | name | description | type | +======+=============================+======+ | from | offset | int | +------+-----------------------------+------+ | size | amount of batches to return | int | +------+-----------------------------+------+ Response Body ^^^^^^^^^^^^^ +------+-----------------------+-----------------+ | name | description | type | +======+=======================+=================+ | id | The batch id | int | +------+-----------------------+-----------------+ | from | offset | int | +------+-----------------------+-----------------+ | size | total amount of lines | int | +------+-----------------------+-----------------+ | log | The log lines | list of strings | +------+-----------------------+-----------------+ REST Objects ============ Session ------- Sessions represent an interactive shell. +----------------+--------------------------------------------------+----------------------------+ | name | description | type | +================+==================================================+============================+ | id | The session id | int | +----------------+--------------------------------------------------+----------------------------+ | kind | session kind (spark, pyspark, or sparkr) | `session kind`_ (required) | +----------------+--------------------------------------------------+----------------------------+ | log | The log lines | list of strings | +----------------+--------------------------------------------------+----------------------------+ | state | The session state | string | +----------------+--------------------------------------------------+----------------------------+ Session State ^^^^^^^^^^^^^ +-------------+----------------------------------+ | name | description | +=============+==================================+ | not_started | session has not been started | +-------------+----------------------------------+ | starting | session is starting | +-------------+----------------------------------+ | idle | session is waiting for input | +-------------+----------------------------------+ | busy | session is executing a statement | +-------------+----------------------------------+ | error | session errored out | +-------------+----------------------------------+ | dead | session has exited | +-------------+----------------------------------+ Session Kind ^^^^^^^^^^^^ +---------+----------------------------------+ | name | description | +=========+==================================+ | spark | interactive scala/spark session | +---------+----------------------------------+ | pyspark | interactive python/spark session | +---------+----------------------------------+ | sparkr | interactive R/spark session | +---------+----------------------------------+ Statement --------- Statements represent the result of an execution statement. +--------+----------------------+---------------------+ | name | description | type | +========+======================+=====================+ | id | The statement id | integer | +--------+----------------------+---------------------+ | state | The execution state | `statement state`_ | +--------+----------------------+---------------------+ | output | The execution output | `statement output`_ | +--------+----------------------+---------------------+ Statement State ^^^^^^^^^^^^^^^ +-----------+----------------------------------+ | name | description | +===========+==================================+ | running | Statement is currently executing | +-----------+----------------------------------+ | available | Statement has a ready response | +-----------+----------------------------------+ | error | Statement failed | +-----------+----------------------------------+ Statement Output ^^^^^^^^^^^^^^^^ +-----------------+-------------------+----------------------------------+ | name | description | type | +=================+===================+==================================+ | status | execution status | string | +-----------------+-------------------+----------------------------------+ | execution_count | a monotomically | integer | | | increasing number | | +-----------------+-------------------+----------------------------------+ | data | statement output | an object mapping a mime type to | | | | the result. If the mime type is | | | | ``application/json``, the value | | | | will be a JSON value | +-----------------+-------------------+----------------------------------+ Batch ----- +----------------+--------------------------------------------------+----------------------------+ | name | description | type | +================+==================================================+============================+ | id | The session id | int | +----------------+--------------------------------------------------+----------------------------+ | kind | session kind (spark, pyspark, or sparkr) | `session kind`_ (required) | +----------------+--------------------------------------------------+----------------------------+ | log | The log lines | list of strings | +----------------+--------------------------------------------------+----------------------------+ | state | The session state | string | +----------------+--------------------------------------------------+----------------------------+ License ======= Apache License, Version 2.0 http://www.apache.org/licenses/LICENSE-2.0