hammerlab / spark-json-relay

SparkListener that converts SparkListenerEvents to JSON and forwards them to an external service via RPC.

JsonRelay

JsonRelay is a SparkListener that converts SparkListenerEvents to JSON and forwards them to an external service via RPC.

It is designed to be used with slim, which consumes JsonRelay's emitted events and writes useful statistics about them to Mongo, from whence Spree serves up live-updating web pages.

Usage

With Spark >= 1.5.0 you can simply pass the following flags to your spark-shell and spark-submit commands:

    --packages org.hammerlab:spark-json-relay:2.0.1
    --conf spark.extraListeners=org.apache.spark.JsonRelay

If using earlier versions of Spark, you'll need to first download the JAR:

$ wget https://repo1.maven.org/maven2/org/hammerlab/spark-json-relay/2.0.1/spark-json-relay-2.0.1.jar

Then, pass these flags to your spark-submit or spark-shell commands:

    --driver-class-path spark-json-relay-2.0.1.jar
    --conf spark.extraListeners=org.apache.spark.JsonRelay

That's it!

Two additional flags, --conf spark.slim.{host,port}, specify the location JsonRelay will attempt to connect and send events to.

Implementation

JsonRelay just piggybacks on Spark's JsonProtocol for JSON serialization, with two differences:

1. It adds an appId field to all events; this allows downstream consumers to process events from multiple Spark applications simultaneously / more easily over time.
2. It rolls its own serialization of SparkListenerExecutorMetricsUpdate events, which is omitted from Spark's JsonProtocol in Spark prior to 1.5.0 (cf. SPARK-9036).

Questions?

Please file an issue if you have any questions about or problems using JsonRelay!

spark-json-relay's Issues

Publish a release on spark packages

jackson-annotations.jar not found in spark 1.5.2

I'm trying to install spree and got stucked when trying to load jsonRelay package.

Here is the command I used.

jinxuans-mbp:spark-1.5.2-bin-hadoop2.6 jinxuanwu$ ./bin/spark-shell  --packages org.hammerlab:spark-json-relay:2.0.1 --conf spark.extraListeners=org.apache.spark.JsonRelay

And I got the following error,

Ivy Default Cache set to: /Users/jinxuanwu/.ivy2/cache
The jars for the packages stored in: /Users/jinxuanwu/.ivy2/jars
:: loading settings :: url = jar:file:/Users/jinxuanwu/spark-1.5.2-bin-hadoop2.6/lib/spark-assembly-1.5.2-hadoop2.6.0.jar!/org/apache/ivy/core/settings/ivysettings.xml
org.hammerlab#spark-json-relay added as a dependency
:: resolving dependencies :: org.apache.spark#spark-submit-parent;1.0
    confs: [default]
    found org.hammerlab#spark-json-relay;2.0.1 in central
    found org.json4s#json4s-jackson_2.10;3.2.10 in list
    found org.json4s#json4s-core_2.10;3.2.10 in list
    found org.json4s#json4s-ast_2.10;3.2.10 in list
    found com.thoughtworks.paranamer#paranamer;2.6 in list
    found com.fasterxml.jackson.core#jackson-databind;2.3.1 in local-m2-cache
    found com.fasterxml.jackson.core#jackson-annotations;2.3.0 in local-m2-cache
    found com.fasterxml.jackson.core#jackson-core;2.3.1 in local-m2-cache
    found org.scala-lang#scalap;2.10.4 in central
    found org.scala-lang#scala-compiler;2.10.4 in list
    found org.scala-lang#scala-reflect;2.10.4 in list
downloading https://repo1.maven.org/maven2/org/hammerlab/spark-json-relay/2.0.1/spark-json-relay-2.0.1.jar ...
    [SUCCESSFUL ] org.hammerlab#spark-json-relay;2.0.1!spark-json-relay.jar (116ms)
:: resolution report :: resolve 2187ms :: artifacts dl 128ms
    :: modules in use:
    com.fasterxml.jackson.core#jackson-annotations;2.3.0 from local-m2-cache in [default]
    com.fasterxml.jackson.core#jackson-core;2.3.1 from local-m2-cache in [default]
    com.fasterxml.jackson.core#jackson-databind;2.3.1 from local-m2-cache in [default]
    com.thoughtworks.paranamer#paranamer;2.6 from list in [default]
    org.hammerlab#spark-json-relay;2.0.1 from central in [default]
    org.json4s#json4s-ast_2.10;3.2.10 from list in [default]
    org.json4s#json4s-core_2.10;3.2.10 from list in [default]
    org.json4s#json4s-jackson_2.10;3.2.10 from list in [default]
    org.scala-lang#scala-compiler;2.10.4 from list in [default]
    org.scala-lang#scala-reflect;2.10.4 from list in [default]
    org.scala-lang#scalap;2.10.4 from central in [default]
    :: evicted modules:
    org.scala-lang#scalap;2.10.0 by [org.scala-lang#scalap;2.10.4] in [default]
    ---------------------------------------------------------------------
    |                  |            modules            ||   artifacts   |
    |       conf       | number| search|dwnlded|evicted|| number|dwnlded|
    ---------------------------------------------------------------------
    |      default     |   12  |   1   |   1   |   1   ||   11  |   1   |
    ---------------------------------------------------------------------

:: problems summary ::
:::: WARNINGS
        [NOT FOUND  ] com.fasterxml.jackson.core#jackson-annotations;2.3.0!jackson-annotations.jar(bundle) (1ms)

    ==== local-m2-cache: tried

      file:/Users/jinxuanwu/.m2/repository/com/fasterxml/jackson/core/jackson-annotations/2.3.0/jackson-annotations-2.3.0.jar

        ::::::::::::::::::::::::::::::::::::::::::::::

        ::              FAILED DOWNLOADS            ::

        :: ^ see resolution messages for details  ^ ::

        ::::::::::::::::::::::::::::::::::::::::::::::

        :: com.fasterxml.jackson.core#jackson-annotations;2.3.0!jackson-annotations.jar(bundle)

        ::::::::::::::::::::::::::::::::::::::::::::::


:::: ERRORS
    unknown resolver null


:: USE VERBOSE OR DEBUG MESSAGE LEVEL FOR MORE DETAILS
Exception in thread "main" java.lang.RuntimeException: [download failed: com.fasterxml.jackson.core#jackson-annotations;2.3.0!jackson-annotations.jar(bundle)]
    at org.apache.spark.deploy.SparkSubmitUtils$.resolveMavenCoordinates(SparkSubmit.scala:1011)
    at org.apache.spark.deploy.SparkSubmit$.prepareSubmitEnvironment(SparkSubmit.scala:286)
    at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:153)
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:120)
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

Could you please help me with this?

Buffer events in the face of loss of connectivity to slim

It would be nice to make the connection to slim a little more robust to e.g. network hiccups.

Remove defunct server/ module

Add tests for JsonRelay code

Would be good to have tests. Even some basic stuff, like checking that JSON conversion is expected or all events are processed or dropped accordingly.

NoSuchMethodError from JsonRelay

Hi there - I'm running json-relay/slim/spree stack against a spark 2.0 cluster in standalone mode. I'm executing scripts through jupyterhub notebooks (pyspark-shell process) and have been receiving this error causing all scripts to crash.

In the pyspark submit args, i'm following the example here with:

--packages org.hammerlab:spark-json-relay:2.0.1
--conf spark.extraListeners=org.apache.spark.JsonRelay

I am able to create the sparkContext but it will crash as soon as I create and perform an action on a dataframe (in this instance, running a SQLQuery):

spark = SparkSession\
    .builder\
    .appName("test01")\
    .getOrCreate()

a = SQLQuery(spark, 'schema', 'user', SQL_BASIC_INFO)
df_users = a.fetch_sql()
df_users.persist()

Stack trace below

16/09/30 22:19:10 ERROR util.Utils: uncaught error in thread SparkListenerBus, stopping SparkContext
java.lang.NoSuchMethodError: org.apache.spark.scheduler.SparkListenerExecutorMetricsUpdate.taskMetrics()Lscala/collection/Seq;
    at org.apache.spark.JsonRelay.onEvent(JsonRelay.scala:51)
    at org.apache.spark.SparkFirehoseListener.onExecutorMetricsUpdate(SparkFirehoseListener.java:103)
    at org.apache.spark.scheduler.SparkListenerBus$class.doPostEvent(SparkListenerBus.scala:59)
    at org.apache.spark.scheduler.LiveListenerBus.doPostEvent(LiveListenerBus.scala:35)
    at org.apache.spark.scheduler.LiveListenerBus.doPostEvent(LiveListenerBus.scala:35)
    at org.apache.spark.util.ListenerBus$class.postToAll(ListenerBus.scala:63)
    at org.apache.spark.scheduler.LiveListenerBus.postToAll(LiveListenerBus.scala:35)
    at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(LiveListenerBus.scala:81)
    at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(LiveListenerBus.scala:66)
    at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1$$anonfun$apply$mcV$sp$1.apply(LiveListenerBus.scala:66)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
    at org.apache.spark.scheduler.LiveListenerBus$$anon$1$$anonfun$run$1.apply$mcV$sp(LiveListenerBus.scala:65)
    at org.apache.spark.util.Utils$.tryOrStopSparkContext(Utils.scala:1229)
    at org.apache.spark.scheduler.LiveListenerBus$$anon$1.run(LiveListenerBus.scala:64)

Hopefully someone can provide some insight into why this is crashing the sparkcontext. I would love to get this stack up and running in a stable way.

Don't make one request per event

It seems the current behavior is likely to become a bottleneck at any reasonable load.

It would be nice to just dump data on an open socket and let the OS do its own queueing/batching.

Publish a JAR to Maven

Work with Spark 1.5.0

Spark 1.5.0 introduced a new SparkListenerEvent, SparkListenerBlockUpdated, which JsonRelay currently MatchErrors on.