Feb 23, 2020 / STREAM PROCESSING, JET, SOURCES, SINKS

Stream processing: sources and sinks

Last week, we had a look at how stream processing could help us compute mathematical approximations using Hazelcast Jet. We described the Pipeline API, and mentioned that it "read" from a source and "wrote" to a sink. This week, I’d like to detail those concepts, what out-of-the-box sinks and sources are available in Jet, and how to create your own should the need be.

This is the 3^rd post in the Stream Processing focus series.Other posts include:

Introduction to stream processing
Stream processing for computing approximations
Stream processing: sources and sinks (this post)

Sources and sinks

I think that most of us are familiar with the batch model. In big companies, it’s been used for ages to read from a database, transform the data, and write to another database: this is known as ETL.

However, databases are not the only place to read from and write to. One other legacy but widespread use-case is to offer a FTP server for other parties to store files to. Then, a daemon process watches if files matching a specific pattern is present: if at least one is found, it reads and process the data, and proceeds to rename/move/delete the file.

In Java EE environments, data can also be read from/written to JMS queues. In that regard, Hazelcast IMDG is also an option.

Source and sink are terms that are broad enough to encompass the diversity of components possible. Those terms apply to both the batching model and the streaming model.

Jet sources and sinks

Out-of-the-box, Jet provides a set of sources and sinks. They are available respectively in the com.hazelcast.jet.pipeline.Sources and com.hazelcast.jet.pipeline.Sinks classes.

Some of them are related to Hazelcast IMDG, most are not:

Source Sink

JMS

Queue

Topic

SQL Database

Query

Statement

Hazelcast

Local/remote Cache

Local/remote IList

Local/remote IMap

Folder

Socket

Standard out

Custom sources

One is not limited to the out-of-the-box provided sources and sinks only. If the desired source/sink is missing, it’s possible to create one’s own using provided abstractions.

Here’s the model relevant to design a custom source:

Jet’s custom source builder class diagram'

batch() is for a bounded collection of items, stream() for an unbounded one.

A sample custom source

The web has a lot of data, mostly available through web services. A scenario where one repeatedly sends requests to the same endpoint to get data is very frequent. Let’s create such a custom Source, using Fuel, a Kotlin library for REST calls.

REST call

fun remoteService(url: String) = SourceBuilder
  .stream("http-source", DummyContext())                                 (1)
  .fillBufferFn(RestSource(url))
  .build()

class RestSource(private val url: String) : BiConsumerEx<Unit, SourceBuffer<String>> {
  override fun acceptEx(unit: Unit, buffer: SourceBuffer<String>) {        (2)
    val (_, _, result) = Fuel.get(url).header().responseString()         (3)
    when (result) {
      is Result.Failure -> println(result.getException())
      is Result.Success -> buffer.add(result.get())                       (4)
    }
  }
}

class DummyContext : FunctionEx<Processor.Context, Unit> {               (1)
    override fun applyEx(ctx: Processor.Context) {}
}

1	Create a stream using a context. The context can be mutable, so as to store state during the iterations. Here, we don’t need to pass anything, so we have an dummy context
2	The function accepts two parameters: the first is the context, the second is a `SourceBuffer` to append to. The buffer will be used by Jet to get the data from
3	Execute the request and assign the result into multiple variables - this is Kotlin’s destructuring declaration in action
4	If the call executed successfully, add the result to the buffer

Using a delay

A lot of publicly available web services enforce some kind of access restriction: it can relate to authentication, rate limiting, etc.

Imagine we need to restrict calls to comply with a rate limiting policy: to avoid getting 429 HTTP status responses, we should wait until a certain time has elapsed before executing the next request. This is the time when an object storing the state becomes necessary. Because Java 8 Date/Time API only provides immutable objects, let’s design a mutable object around it:

class TimeHolder : Serializable {
  var value : Instant = Instant.now()                    (1)
    private set                                          (2)
  fun reset() {
    value = Instant.now()                                (3)
  }
}

1	The initial instant is set to now
2	Enforce encapsulation so that only the class itself can set `value`
3	Reset `value` to now

The context creation function needs to be updated accordingly:

class CreateContext : FunctionEx<Processor.Context, TimeHolder> {
  override fun applyEx(ctx: Processor.Context) = TimeHolder()
}

At this point, the wait logic can be used in the RestSource:

class RestSource(private val url: String) :
    BiConsumerEx<TimeHolder, SourceBuilder.SourceBuffer<String>> {
  override fun acceptEx(time: TimeHolder, buffer: SourceBuilder.SourceBuffer<String>) {
    if (Instant.now().isAfter(time.value.plusSeconds(30))) {          (1)
      val (_, _, result) = Fuel.get(url).header().responseString()
      when (result) {
        is Result.Failure -> println(result.getException())
        is Result.Success -> buffer.add(result.get())
      }
      time.reset()                                                    (2)
    }
  }
}

1	Create a guard to check whether the time elapsed is at least 30 seconds after the previous request
2	Store the current time to be available for the next check

Conclusion

Hazelcast Jet provides some sources and sinks out-of-the-box for most common use-cases. In addition, it also offers a low-level API to create one’s own in case none fits one’s needs.

To go further:

Follow @nicolas_frankel