spark-http-stream

spark-http-stream transfers Spark structured stream over HTTP protocol. Unlike tcp streams, Kafka streams and HDFS file streams, http streams often flow across distributed big data clusters on the Web. This feature is very helpful to build global data processing pipelines across different data centers (scientific research institues, for example) who own seperated data sets.

spark-http-stream provides:

HttpStreamServer: a HTTP server which receives, collects and provides http streams
HttpStreamSource: reads messages from a HttpStreamServer, acts as a structured streaming Source
HttpStreamSink: sends messages to a HttpStreamServer using HTTP-POST commands, acts as a structured streaming Sink

also spark-http-stream provides:

HttpStreamClient: a client used to communicate with a HttpStreamServer, developped upon HttpClient
HttpStreamSourceProvider: a StreamSourceProvider which creates HttpStreamSource
HttpStreamSinkProvider: a StreamSinkProvider which creates HttpStreamSink

The simple archtecture of spark-http-stream is shown below:

importing spark-http-stream

use maven to import spark-http-stream:

<dependency>
    <groupId>org.grapheco</groupId>
    <artifactId>spark-http-stream</artifactId>
    <version>0.9.1</version>
</dependency>

Starts a standalone HttpStreamServer

HttpStreamServer is actually a Jetty server with a HttpStreamServlet, it can be started using following code:

val server = HttpStreamServer.start("/xxxx", 8080);

When http://localhost:8080/xxxx is requested, the HttpStreamServlet will use an embeded ActionsHandler to parse request message, perform certain action(fecthSchema, fetchStream, etc), and return response message.

By default, an NullActionsHandler is provided. Of coz it can be replaced with a MemoryBufferAsReceiver:

server.withBuffer()
	.addListener(new ObjectArrayPrinter())
	.createTopic[(String, Int, Boolean, Float, Double, Long, Byte)]("topic-1")
	.createTopic[String]("topic-2");

or with a KafkaAsReceiver:

server.withKafka("vm105:9092,vm106:9092,vm107:9092,vm181:9092,vm182:9092")
	.addListener(new ObjectArrayPrinter());

as shown above, several kinds of ActionsHandler are defined in spark-http-stream:

NullActionsHandler: does nothing
MemoryBufferAsReceiver: maintains a local memory buffer, stores data sent from producers into buffer, and allows consumers to fetch data in batch
KafkaAsReceiver: forwards all received data to Kafka

Notes that MemoryBufferAsReceiver maintains a server-side message buffer, while KafkaAsReceiver only forwards messages to Kafka cluster.

HttpStreamSource, HttpStreamSink

The following code shows how to load messages from a HttpStreamSource:

val lines = spark.readStream.format(classOf[HttpStreamSourceProvider].getName)
	.option("httpServletUrl", "http://localhost:8080/xxxx")
	.option("topic", "topic-1");
	.option("includesTimestamp", "true")
	.load();

options:

httpServletUrl: path to the servlet
topic: topic name of messages to be consumed
includesTimestamp: tells if each row in the loaded DataFrame includes a time stamp or not, default value is false
timestampColumnName: name assigned to the time stamp column, default value is '_TIMESTAMP_'
msFetchPeriod: time interval in milliseconds for message polling, default value is 1(1ms)

The following code shows how to output messages to a HttpStreamSink:

val query = lines.writeStream
	.format(classOf[HttpStreamSinkProvider].getName)
	.option("httpServletUrl", "http://localhost:8080/xxxx")
	.option("topic", "topic-1")
	.start();

options:

httpServletUrl: path to the servlet
topic: topic name of produced messages
maxPacketSize: max size in bytes of each message packet, if the actual DataFrame is too large, it will be splitted into serveral packets, default value is 10*1024*1024(10M)

Note that HttpStreamSource is only available when the HttpStreamServer is equiped with a MemoryBufferAsReceiver (use withBuffer, as shown above). If the HttpStreamServer choose Kafka as back-end message system (use withKafka), it is wrong to consume data from HttpStreamSource, just use KafkaSource (see http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html) instead:

val df = spark
	.readStream
	.format("kafka")
	.option("kafka.bootstrap.servers", "vm105:9092,vm106:9092,vm107:9092,vm181:9092,vm182:9092")
	.option("subscribe", "topic-1")
	.load()

see https://github.com/bluejoe2008/spark-http-stream/blob/master/src/test/scala/HttpStreamSourceSinkTest.scala and https://github.com/bluejoe2008/spark-http-stream/blob/master/src/test/scala/HttpStreamKafkaTest.scala to get complete example code.

Understanding ActionsHandler

as shown in previous section, serveral kinds of ActionsHandler are defined in spark-http-stream: NullActionsHandler, MemoryBufferAsReceiver, KafkaAsReceiver.

users can also customize their own ActionsHandler as they will. The interface looks like:

trait ActionsHandler {
	def listActionHandlerEntries(requestBody: Map[String, Any]): ActionHandlerEntries;
	def destroy();
}

here ActionHandlerEntries is just an alias of PartialFunction[String, Map[String, Any]], which accepts an input argument action: String, and returns an output argument responseBody: Map[String, Any]. the listActionHandlerEntries method is often written as a set of case expression:

override def listActionHandlerEntries(requestBody: Map[String, Any])
	: PartialFunction[String, Map[String, Any]] = {
	case "actionSendStream" ⇒ handleSendStream(requestBody);
}

the code shown above says: this ActionsHandler only handles action actionSendStream, in this case, it calls the method handleSendStream(requestBody) to handle request and output its return value as response. If other action is requested, an UnsupportedActionException will be thrown by the HttpStreamServer.

ActionsHandlerFactory is defined to tell how to create a ActionsHandler with required parameters:

trait ActionsHandlerFactory {
	def createInstance(params: Params): ActionsHandler;
}

Embedding HttpStreamServer in Web application servers

spark-http-stream provides a servlet named ConfigurableHttpStreamingServlet, users can configure the servlet in web.xml:

<servlet>
	<servlet-name>httpStreamServlet</servlet-name>
	<servlet-class>org.apache.spark.sql.execution.streaming.http.ConfigurableHttpStreamServlet</servlet-class>
	<init-param>
		<param-name>handlerFactoryName</param-name>
		<param-value>org.apache.spark.sql.execution.streaming.http.KafkaAsReceiverFactory</param-value>
	</init-param>
	<init-param>
		<param-name>bootstrapServers</param-name>
		<param-value>vm105:9092,vm106:9092,vm107:9092,vm181:9092,vm182:9092</param-value>
	</init-param>
</servlet>

<servlet-mapping>
	<servlet-name>httpStreamServlet</servlet-name>
	<url-pattern>/xxxx</url-pattern>
</servlet-mapping>

As shown above, a servlet of ConfigurableHttpStreamServlet is defined with a ActionsHandlerFactory KafkaAsReceiverFactory, required parameters for the ActionsHandlerFactory (bootstrapServers, for example), are defined as a set of init-params.

Using HttpStreamClient

HttpStreamClientprovides a HTTP client used to communicate with aHttpStreamServer`. It contains serveral methods:

sendDataFrame: send a DataFrame to the server, if the DataFrame is too large, it will be splitted into smaller packets
sendRows: send data (as Array[Row]) to server
fetchSchema: retrieves schema of certain topic
fecthStream: retrieves data (as 'Array[RowEx]') from server
subscribe: subscribe a topic and retrieves a subscriberId
unsubscribe: unsubscribe

Note that some methods are only available when the server is equipped with correct ActionsHandler. As an example, the KafkaAsReceiver only handles action actionSendStream, that means, if you called fetchStream and sendDataFrame methods of the HttpStreamClient, it works well. But it will fail and throw an UnsupportedActionException when you called subscribe method.

+---------------+------------------------+-----------------+
|  methods      | MemoryBufferAsReceiver | KafkaAsReceiver |
+---------------+------------------------+-----------------+
| sendDataFrame |             √          |        √        |
+---------------+------------------------+-----------------+
| sendRows      |             √          |        √        |
+---------------+------------------------+-----------------+
| fetchSchema   |             √          |        X        |
+---------------+------------------------+-----------------+
| fecthStream   |             √          |        X        |
+---------------+------------------------+-----------------+
| subscribe     |             √          |        X        |
+---------------+------------------------+-----------------+
| unsubscribe   |             √          |        X        |
+---------------+------------------------+-----------------+

StreamListener

StreamListener works when new data is arrived and will be consumed by ActionsHandler:

trait StreamListener {
	def onArrive(topic: String, objects: Array[RowEx]);
}

Two kinds of StreamListeners are provided:

StreamCollector: collects data in a local memory buffer
StreamPrinter: prints data while arriving

an example messages look like this:

++++++++topic=topic-1++++++++
RowEx([hello1,1,true,0.1,0.1,1,49],1,0,2017-08-27 20:37:56.432)
RowEx([hello2,2,false,0.2,0.2,2,50],1,1,2017-08-27 20:37:56.432)
RowEx([hello3,3,true,0.3,0.3,3,51],1,2,2017-08-27 20:37:56.432)

Schema, data types, RowEx

spark-http-stream only supports data types which can be recognized by Spark Encoders. These data types includes: String, Boolean, Int, Long, Float, Double, Byte, Array[].

A row will be wrapped as a RowEx object on receiving. RowEx is a data structure richer than Row. It contains some members and methods:

originalRow: original row
batchId: batch id passed by Spark
offsetInBatch: offset of this row in current batch
withTimestamp(): returns a Row with a timestamp
withId(): returns a Row with its id
extra(): returns a triple (batchId, offsetInBatch, timestamp)

Considering an original row has values [hello1,1,true,0.1,0.1,1,49], following code show contents of mentioned structures:

originalRow:

+---------------+-------+--------------+-----------+------------+--------+---------+
| String:hello1 | Int:1 | Boolean:true | Float:0.1 | Double:0.1 | Long:1 | Byte:49 | 
+---------------+-------+--------------+-----------+------------+--------+---------+

RowEx:

+---------------+-------+--------------+-----++--------+-------+-------------------------------+
| String:hello1 | Int:1 | Boolean:true | ... || Long:1 | Int:0 | Timestamp:2017-08-27 20:37:56 |
+---------------+-------+--------------+-----++--------+-------+-------------------------------+

withTimestamp():

+---------------+-------+--------------+-----------+-----+-------------------------------+
| String:hello1 | Int:1 | Boolean:true | Float:0.1 | ... | Timestamp:2017-08-27 20:37:56 |
+---------------+-------+--------------+-----------+-----+-------------------------------+

withId():

+---------------+-------+--------------+-----------+------------+--------+---------+------------+
| String:hello1 | Int:1 | Boolean:true | Float:0.1 | Double:0.1 | Long:1 | Byte:49 | String:1-0 |
+---------------+-------+--------------+-----------+------------+--------+---------+------------+

extra():

+--------+-------+-------------------------------+
| Long:1 | Int:0 | Timestamp:2017-08-27 20:37:56 |
+--------+-------+-------------------------------+

SerDe

spark-http-stream defines a SerilizerFactory to create a SerializerInstance:

trait SerializerFactory {
	def getSerializerInstance(serializerName: String): SerializerInstance;
}

an SerializerFactory.DEFAULT object is provided which is able to create two kinds of serializers:

java: creates a JavaSerializer
kryo: creates a KryoSerializer

New kind of Serializer, json serializer, for example, is welcome.

By default, HttpStreamClient and HttpStreamServer uses kryo serializer.

Tests

HttpStreamServerClientTest: tests HttpStreamServer/Client, https://github.com/bluejoe2008/spark-http-stream/blob/master/src/test/scala/HttpStreamServerClientTest.scala
HttpStreamSourceSinkTest: tests HttpStreamSource and HttpStreamSink, https://github.com/bluejoe2008/spark-http-stream/blob/master/src/test/scala/HttpStreamSourceSinkTest.scala
HttpStreamKafkaTest: tests HttpStreamSink with Kafka as underlying message reveiver, https://github.com/bluejoe2008/spark-http-stream/blob/master/src/test/scala/HttpStreamKafkaTest.scala
HttpStreamDemo: a tool helps to test HttpTextStream and HttpTextSink, https://github.com/bluejoe2008/spark-http-stream/blob/master/src/test/scala/HttpStreamDemo.scala

steps to tests HttpStreamDemo:

choose machine A, run HttpStreamDemo start-server-on 8080 /xxxx, this starts a HTTP server which receives data from machine B
choose machine B, run nc -lk 9999
run HttpStreamDemo read-from http://machine-a-host:8080/xxxx on machine B
run HttpStreamDemo write-into http://machine-a-host:8080/xxxx on machine C
type some text in nc, data will be received by HttpStreamSink and then consumed as HttpStreamSource, finally displayed on console

dependencies

kafka-clients-0.10: used by KafkaAsReceiver
httpclient-4.5: HttpStreamClient uses HttpClient project
jetty-9.0: HttpStreamServer is devploped upon Jetty
spark-2.1: spark structued streaming libray

Cheap and reliable Node.js hosting starts at $3/month, and $1/month static HTML hosting

bluejoe2008 / spark-http-stream

Programming Languages

Labels

Projects that are alternatives of or similar to spark-http-stream