We describe the evolution of RAM3S, a software infrastructure for the integration of Big Data stream processing platforms, to SPAF, an abstraction framework able to provide programmers with a simple but powerful API to ease the development of stream processing applications. By using SPAF, the programmer can easily implement real-time complex analyses of massive streams on top of a distributed computing infrastructure, able to manage the volume and velocity of (multimedia) Big Data streams.
To exemplify the many RAM3S/SPAF
components throughout the paper, we will use
the first RAM 3S application we created [
Several modern applications (e.g., face detection for the automatic identification of “suspect”
people [
The initial challenge that led us to the introduction of RAM3S was the implementation of a number of security-related MM stream processing applications on top of such SPEs, with the goal of comparing them by way of several performance KPI, such as throughput, scalability, latency, etc. This required to re-implement every MM stream analysis application on top of each SPE, leading to huge code replication and other ineficiencies. We were therefore led to realize a middleware software framework to allow users to avoid having to deal with details of each specific SPE (such as how fault-tolerance is achieved, how stream data are stored, etc.), and easily extend already available (centralized) software to a scaled out solution. This way, we strove to bridge the technological gap between facilities provided by SPEs and advanced applications (whose implementation to a distributed computing scenario could be daunting).
Our original goal for RAM3S was to create a framework, according to the definition in [
The current version of RAM3S allows to experiment with the various SPEs (namely, Spark Streaming, Storm, Flink, and Samza) by providing a separate “generic” application for each of them. Each of such applications allows the execution of a specific Figure 2: RAM3S programming interface and its instantiation for example on the respective face recognition.
SPE. RAM3S interposes an abstraction layer based on interfaces between the generic and the example applications (see Figure 2). Such interfaces model aspects of both data streaming and data processing: The Receiver interface represents the external system from which the application receives data; its receive method accommodates the logic of receiving the single processable object from the external system. The Analyzer interface represents the container of all the processing logic of the application; its analyze methods takes a MM object as input, generating a single object as result. Finally, the ApplicationFactory interface is responsible for representing the application as a unit; in essence, it serves as a collector for the Analyzer implementation and for the Receiver implementation, by instantiating the concrete Analyzer and Receiver type classes defined in the application context.
Let us now consider how RAM3S interfaces public static void main(String[] args) { are used by a generic application. The purpose ApplicaSteicounrFiatcytFoarcytofrayc.tcolrayss=.newInstance(); of such application is to map the application RAencaeliyvzeerr raencaeliyvzeerr == ffaaccttoorryy..ccrreeaatteeRAencaeliyvzeerr(());; (defined in terms of the above interfaces) to /.*..Boilerplate Code specific for the SPE */ the relevant SPE, thus executing its logic on } the underlying runtime framework.
Generic application code is always com- Figure 3: Code of a generic RAM3S application. pletely specified within the main method and has a recurring structure, presented in Figure 3. The concrete factory is used to create the Receiver and Analyzer; then such objects have to be translated into objects and procedures specific of the underlying SPE, as exposed by its APIs; the purpose of this step is to concretely establish the connection to the specific data processing infrastructure and to implement the application logic. Clearly, this is so-called boilerplate code, peculiar to the underlying SPE, that has to be repeated, almost verbatim, for each specific application.
A final component of RAM 3S is the one used to support diferent message brokers [
From the analysis we have presented, we can conclude that, at present, RAM3S allows diferent stream processing applications to be executed in a facilitated manner, but is not yet able to allow the definition of a new application without having to write part of the code of RAM3S itself. This drawback places RAM3S in the role of a “quasi-framework”. On the other hand, the message broker support is independent of the application code (thus satisfying the reusability requirement) and makes it possible to decouple the implementation of the read and write “adapters” of a certain message broker from the underlying stream processing engine (thus also ofering extensibility). In the next section, we will introduce SPAF, whose main goal is to enable the use of RAM3S according to the original intent, namely to facilitate the creation of new stream processing applications, while ofering both reusability and extensibility.
Before describing the concepts that have been used to define our Stream Processing Abstraction Framework, SPAF, it is useful to recollect the original requirements for RAM3S: Facilitate the creation of stream processing applications: the SPAF API should explicitly expose the pivotal concepts of stream processing and, more importantly, allow the application to be defined by writing code as close as possible to a description in natural language. Checking type-safety at compile time would be also helpful. Framework independence: here, the target user is not the programmer of applications, rather the developer who wants to extend SPAF to work with a diferent SPE. Such programming interface is called Service Provider Interface (SPI) and is a programming pattern supported natively by Java. Connector independence: it should be possible to integrate, in a pluggable manner, new connector providers (e.g., message queues, file systems, databases), again exposing a SPI to be implemented by developers. SPAF will therefore expose a dual abstraction layer: one for SPEs and one for input/output supports. Connector abstraction actually concerns the application programmer as well, since the SPAF API should relieve the programmer of the implementation details regarding the use of each connector’s libraries and allow him to specify sources and destinations in a declarative manner.
In the first version of SPAF, a number of simplifying assumptions has been introduced to ease implementation of the previously described requisites: 1. Connectors will be limited to message queues, as provided by well-known message brokers such as Apache Kafka and RabbitMQ. 2. Streams will be composed exclusively by key-value pairs. 3. No support will be provided for storing intermediate computation results, i.e., stream processing will be stateless. 4. The processors operating on streams will accept a single input stream and a single output stream; essentially, it will be possible to define only “linear” topologies. 5. It will be possible to specify only the logical topology (definition of the transformation process from input to output) and not the physical topology, i.e., defining of how various computational elements of the logical topology (sources, transformation operators, destinations) are distributed on the physical nodes.
grammer to configure the execution environment in an abstract way (on the API side) and to implement such configurations in a provider-specific way (on the SPI side).
A Topology defines the computational logic of a stream processing application, that is, how the input data are transformed into the output data. As said, in the first version of SPAF only linear topologies are supported. We conceived the framework to accept specifications of the logic of Processor nodes also via lambda-functions and present a fluent API to compose the topology. To achieve this, we exploited a “creational-type” design pattern called Builder, allowing complex objects (like a topology) to be constructed step by step; this could also re-used in future versions of SPAF, where it may be convenient to change the way the topology is represented (e.g., to support DAG topologies).
The concept of Application basically coincides with the one of Topology, essentially adding descriptive information only. Conceptually, however, a stream processing application could define more than one topology: this is why the two entities are separated, although, in this first version of SPAF, there is a 1-1 relationship between Application and Topology.
Source and Sink clearly denote the source and destination, respectively, for data in a stream processing application, while a Processor represents a node in the Topology, implementing a processing step that is used to transform data, thus realizing the actual data processing logic. Processors can be thought as “black boxes” with a single input and a single output, where data transformation can be defined arbitrarily through the process method. The additional init method can be defined in those cases needing a one-of initialization of the Processor.
Finally, the Element entity represents the only data type that can be streamed in a SPAF Topology. The Element class is somewhat hidden from the application programmer, while its use appears evident in the SPI layer, for both SPEs and connectors.
To create an application in SPAF, it is necessary to follow some steps, mostly independent of the specific streaming application logic to be created. In the following, we will exemplify the creation of an application for the use case of face recognition in videos (see Section 1.1): As it will be clear, the application-specific code can be easily distinguished from the generic SPAF-based code. The code needed for the complete specification of a SPAF application has actually a dual nature: some declarative code, included in a configuration file, needed to specify the customization of SPAF entities, like Context, Source, or Sink, and some procedural code, used to instantiate SPAF classes and to specify the actual stream processing application logic; this is included in the main method of the application and, again, is mainly boilerplate code.
Figure 7 shows, for the specific use case, (left) a possible config file, with specification of the context (using Flink as SPE), of the application, and of the input and output connectors (using Kafka), and (right) the code of the main method of the FaceRecognition application. It is clear that most of the code is indeed boilerplate, i.e., repeated with no variation across diferent applications. The only specific part is at step 3., where we declare that this application is composed of three main phases: 1) detection of faces in each image, 2) recognition of detected faces, and 3) (possible) marking of recognized faces (note that these correspond exactly with the three steps illustrated in Section 1.1). Obviously, the programmer should also write the code for each individual Processor, but this is absolutely independent of the underlying stream context { flink { local = true web-ui = true } web-ui-port = 10081 sink { type = kafka bootstrap-servers = "kafka:9092" } topic = RECOGNIZED_FACES // 1. configuration and creation of the execution environment Config config = ConfigFactory.load();
Context context = StreamProcessing.createContextFactory().createContext(config); } S/o/ur2c.e<dSetfriinnigt,ioSntroifngi>npsuoturacned =ouSttpruetamsPtrroecaemsssing.createSource(config); appnlaimceat=io"nFa{ce Recognition" /S/in3k<.Sdtnreofidniengsi,)tSitorninogf>tshienklo=giSctarleatmoPproolcoegsysi(nig.e..c,reahotewSdiankt(acionnftirg)a;nsformed in processing } dataset-path = "/tmp/training-faces/" Topology topology = new Topology()
.setSource("Source", source) soubtroycopetes{t=rakapf-skaervers = "kafka:9092" ....saaaedddtdddSPPPirrrnoookccc(eee"sssSsssiooorrrnk((("""",FFFaaascccieeenMRDkaee)rct;koeegcrnt"io,zre"n,re"nw,ePnweeFrwasoFcnaeDFceaecRteeecMcaotrgioknniintPrgioPorncoPecrsseooscser(os)rs(o,)r,("Sco"oFnuarfccieegR)"e,)co"gFaniczeeDer"t)ector") } topic = FACES A/p/pl4i.cactrieoantiaopnploifcatthieonap=plniecwatAipopnlication() .withName(config.getString("application.name")) .withTopology(topology); // 5. launch of the application context.run(application); processing infrastructure (following the original goal of RAM3S). Step 3. also demonstrates the use of the Builder pattern, where each Processor refers to its predecessor through its id.
A dimension of fundamental importance, which should be taken into account when choosing any development tool, is the so-called “learning curve”, which relates the level of knowledge and the time invested in learning a new thing. SPAF plays the role of the “guide” that takes the programmer through the discovery of stream processing concepts, providing a logical path that facilitates their understanding and thus making the learning curve of stream processing less daunting. SPAF is therefore able to make life easier for the inexperienced programmer who must venture into the world of stream processing for the first time, and have that journey less treacherous (like Virgil, accompanying Dante through the “hell” of stream processing).
In the previous section, where we described the main SPAF concepts and illustrated how they are used to implement a specific stream processing application, we deliberately omitted several issues that have been tackled in the design and implementation of SPAF, like the relationship between SPEs and connectors, or the one between type-safety and serialization.
Additional details on the SPAF technicalities can be found in [
When considering the representation of topologies in SPAF, we should remind that the key operation for a topology is how the SPAF stream processing layer (see Figure 5) is able to map the Topology entity in the corresponding topology of the underlying SPE (who will then autonomously map it to a physical topology on computing nodes). Since this version of SPAF only accepts linear topologies, this mapping is extremely simple, being implemented as a loop visiting all topology nodes in an ordered way. Refactoring the Source, Processor, and Sink entities (Figure 6) currently used to build topologies, an abstract entity TopologyNode, exposing a common interface, could be created. In this way, Sources, Processors, and Sinks could be considered as TopologyNode instances, allowing to use polymorphism to implement simple (and elegant) algorithms for mapping topologies, e.g., by exploiting a Visitor pattern based on a topological sort of the DAG.
Finally, we consider the use of multiple SPAF applications, in a simultaneous and coordinated manner, with the aim of solving a stream processing problem in a “highly distributed” way (we will soon clarify what we mean by this adjective). The main idea is based on the decomposition of the stream processing problem into sub-problems, and in solving them through the use of multiple, independent but cooperative, SPAF projects.
Each SPAF application will define a certain logical topology able of solve a certain subproblem; each topology will receive data via the connectors provided by SPAF, process Figure 9: Super-topologies in SPAF. them, and send the processed data back to the outside world. In this scenario, message queues are used as means of communication between the topologies of individual projects. In other words, we can implement a topology of topologies (see Figure 9). Individual topologies, in fact, can be thought as “black boxes”, processing nodes of a DAG, receiving and sending data via the arcs connecting them, the latter realized by diferent message queues. What we just described corresponds precisely to the definition of topology given in Section 3.1, but at a higher level, thus the name of super-topology, where the prefix super- is to be understood in the Latin sense of “that stands above”.
In the diagram of Figure 9, it is shown how to conceptually realize a SPAF super-topology. It will be necessary to provide “border” sources and sinks (shown in blue and yellow, respectively), functioning as the input and outputs of the entire super-topology. The processing (red) nodes of the super-topology will instead correspond to a single SPAF application each, defining its own Source and Sink, and will consist of a (currently, only linear) Topology of Processors. Communication between the nodes of the super-topology, i.e., between SPAF applications, is realized by way of appropriate message queues between node pairs (these are represented by red arcs in the figure and exemplified on the right). Obviously, each topology will have to be configured to receive and send data to the right message queues, whether they are “border” (in the example, topologies T1, T2, T4 and T5) or “internal” (T3). The use of super-topologies in SPAF opens up some interesting scenarios: 1. each SPAF application, i.e., each node in the super-topology, can be executed by a potentially diferent SPE, since each application can in fact specify the desired SPAF provider independently from the others; 2. it follows that each SPAF application, i.e., each node of the super-topology, can potentially be executed on a cluster of nodes by itself; for this, we previously used the term “highly distributed” execution. The authors thank Nicolò Scarpa for implementing SPAF and maintaining the GitHub repository.1 Most of the figures, as well as the analogy between SPAF and Virgil, have been created by Nicolò, and elaborated by the authors.