Edge-Stream: A Stream Processing Approach For Distributed .

&ORXG(a) IoT-Edge-Cloud(c) IoT-Edge-IoTto the result. To the best of our knowledge, there are seldomframeworks that are able to provide range-based data aggregation. For example, EdgeX [10] and AWS Greengrass [3]allow users to create topics to collect data from sources andlet sinks to subscribe. However, those topics are mostly definedin advance, instead of dynamically following the data sink. If avehicle wants to know the number of cars around it, it is hardto select a proper topic to accumulate. A detailed descriptionof the case is introduced in Subsection III-D.The last type called IoT-Edge focuses on a special scenario,in which data processing results are collected by the same datasource device. Such case is common in mobile applications,such as gaming, VR, and AR. If the IoT or mobile devicesare not able to complete the computation all by themselves,they have to borrow computing resources from other powerfulnodes nearby. Cloudlet [32] allows devices to discover nearbyCloudlet servers to help them with the workload. MobileFog [18] proposes a Platform-as-a-Service (PaaS) model toprovide developers with a simplified programming interface.Paradrop [39] serves the endpoint users with a multi-tenantplatform. Chen et al. [8] demonstrates the multi-user computation offloading procedure as a game. The scheduling problemis transformed into reaching a Nash Equilibrium.&ORXG(b) Cloud-Edge-IoT(d) IoT-EdgeFig. 2: Four typical scenarios of IoT applicationsto describe their dataflow graph in a high-level programmaticwiring language on a sensor network, based on the functionallanguage OCaml. Couper [19] focuses on deploying deeplearning algorithms on a three-tier edge computing system.It supports quick creation of slices of production DNNs forvisual analytics, and enables their deployment in contemporarycontainer-based edge software stacks.C. Motivation of Our ApproachThe second scenario is called Cloud-Edge-IoT, as illustrated in Figure 2 (b). It is common for applications likelive-streaming. In this case, data are pushed down from datacenters towards a large group of endpoint devices. The keyidea is to cache the hot content at the edge of the network, andthus reduce the transmission delay of contents and relieve thebandwidth pressure for data-centers. The content delivery network [36] has already become a mature approach for Internetbased caching by deploying cache servers at the edge ofInternet. Information-centric networking [1] provides contentdistribution services to mobile users with a wireless cacheinfrastructure. Femtocaching [13] studies a wireless distributedcaching network. It assists the macro base station by handlingrequests of popular files that have been cached. Xing et al. [40]proposes a distributed multi-level storage model to providestorage and computing services for IoT devices with limitedstorage capacity and poor network stability.From the discussion in the last subsection, we can find thatmost edge-computing applications can be categorized into oneaforementioned scenario or a combination of multiple ones,such as a complex smart city system [44]. Unfortunately, moststate-of-art edge-computing systems/infrastructures only focuson a single scenario. There still lacks a unified programmingmodel that can support all the scenarios. In addition, the lowlevel design details are not well hidden in some of edgecomputing systems. Moreover, special hardware support isrequired for some of edge-computing infrastructures, such asAmazon AWS Greengrass [3], Microsoft Azure IoT Hub [28],and Google Cloud IoT Edge [14]. Thus, the development ofedge computing on a real-world application is inefficient, especially when multiple scenarios are involved in one application.Traditional distributed computation frameworks in datacenters, including Flink [6] and Storm [35], are designedto deal with continuous data packages generated from IoTdevices, web applications and social media. The stream processing model is adopted to describe the jobs. The endlesssequence of data packages of the same type are abstracted as astream. Developers apply a series of operators to process eachelement in the stream. Such high-level abstraction is friendlyto programmers, and is expressive to describe different typesof jobs in data-centers.Inspired by the method, we introduce a new stream processing model, Edge-Stream, for edge computing scenarios. Itis rather simple and easy to program, but is general enoughto cover the typical scenarios mentioned above. At the sametime, we implement a prototype of edge-computing frameworkbased on Edge-Stream model on commodity hardware. Detailsare introduced in following sections.The third scenario is named as IoT-Edge-IoT in this work,as shown in Figure 2 (c). It refers to those applications thatboth data producers and consumers are endpoint IoT devices.When tracing the data-flow, we can find that data requestsusually come from devices, which are close to the data sourcedevices geographically. A straightforward method of edgeprocessing for this scenario relies on an interconnected localnetwork, such as the Wireless Sensor Networks (WSN) [30].DDF [12] introduces an approach to develop programs usinga dataflow model. This scenario is common for applications,such as smart transportation and smart manufacturing. In fact,Amazon AWS Greengrass [3], Microsoft Azure IoT Hub [28]and Google Cloud IoT Edge [14] have provided their architectures, where sensors collect data to analyze and actors react Authorized licensed use limited to: University of Minnesota. Downloaded on August 11,2021 at 14:16:56 UTC from IEEE Xplore. Restrictions apply.

sFilesLicensePlate rayCamerasRecogFig. 4: An analogy with Linux shell script exampleSaveThe abstract view of this example is illustrated in Figure 3 (b). It contains three streams and two operators. In eachabstract view, we put operators on the top and allocate allstreams in the bottom. In the rest of this paper, we will useabstract views to introduce more examples. The correspondingcode based on Edge-Stream model is presented in Listing 1.The first line declares two operators (i.e. Recognition andSave). The second line imports the video stream. License platestream and storage stream are generated using operators in lasttwo lines, respectively.Data-sequenceStreamsVideo StreamLicense PlateStreamresult.txtcat raw.txt grep "hello" result.txt(a) Physical system illustrationOperatorspipe fileStorage Stream(b) Abstract view of the jobFig. 3: Different views for a license plate recognition jobIII. E DGE -S TREAM M ODELrequire Recog, Saveimport VideoStreamPlateStream Recog VideoStreamStorageStream Save PlateStreamIn this section, we mainly introduce the components of theEdge-Stream model. First, we provide a case study in orderto facilitate the explanation. Then we describe the streamabstraction from the perspective of each basic components.As the Edge-Stream model borrows some concepts from thetraditional file system, we also compare some coincident partbetween them, and emphasize the specific features for streams.Listing 1: Code for the license plate recognition jobTo provide better understanding of these components inEdge-Stream model, we use the file system as a close analogy.A stream in an edge system can be analogous to a file in a filesystem. A data-sequence can be analogous to a line in eachfile. The operators are analogous to shell commands, such astail, grep, cat.We also provide an example of manipulating files via Linuxshell commands, which is shown in Figure 4, to demonstratethe similarity. Command cat reads each line in raw.txt, andpass on them with a pipe. The pipe process will create anew pipe file in the PipeFS, a special file system mountedin the kernel. Then command grep reads each line in thepipe file, and writes the result to result.txt. It is easy to tellthat the example of manipulating file in Figure 4 is quitesimilar to that of programming with Edge-Stream model inFigure 3 (b). Design details of “file (stream)” and “shellcommands (operators)” in Edge-Stream model are introducedin the following subsections.A. Basic ComponentsWhen we use Edge-Stream model to describe a scenario, itcontains several basic components, including “data-sequence”,“stream”, and “operator”. Their definitions are introduced asfollows using a simple example of intelligent traffic systemillustrated in Figure 3.a) Data-sequence: Data-sequence is the basic unit thatcan be manipulated by an operator in the model. As shownin Figure 3 (a), each camera generates a data-sequence ofcaptured vehicles. These data-sequences are used for dataanalysis (e.g. license plate recognition) later in the system.b) Stream: Stream is a set of data-sequences, which areprocessed with the same operator. As shown in Figure 3 (a),there are four cameras in total. All data-sequences from thesecameras are grouped together as a video stream.c) Operator: Operator is a user-defined function in anedge system. It takes one or multiple streams as the inputand generates a single stream as the output. For the examplein Figure 3 (a), there are two operators. The first operator isLicense Plate Recognition (Recog). It takes the video streamfrom cameras as input and generates a license plate stream,which contains the license plates of vehicles. The secondoperator is Save. It takes the license plate stream as the inputand turns it into a new stream for storage.B. Stream DesignBesides a set of data-sequences, a stream also maintains alist of metadata indicating its properties. Operators need tocheck those metadata items to determine whether the streammeets the requirements as an input. The key items of thestream metadata are listed in Table I. Other minor features,such as the stream name, are omitted from the list. Authorized licensed use limited to: University of Minnesota. Downloaded on August 11,2021 at 14:16:56 UTC from IEEE Xplore. Restrictions apply.

TABLE I: Basic metadata items of a streamMetadataT ypeOwnerW indowSerializerMeaningClassifying the stream into several categories:primitive, virtual, generated.Indicating the user that creates and possessesthe stream.Describing the window type of the stream.Indicating the serializing method.TypeWindowSerializerMetadata Stream Type Primitive /Type SourcePool Source 192.168.10.* /Source /SourcePool /Stream Stream Type Virtual /Type SourcePool Source date %M %S /Source /SourcePool /Stream Listing 2: Stream declaration examplesPrimitive streams are the original sources of data in thesystem. There are basically two kinds of them: device-basedand range-based. They get data-sequences from a list ofphysical devices, and a list of areas, respectively. For example,stream owner X has deployed a bunch of sensors, and is able toget data-sequence from them. Thus it is device-based stream.An other case is that stream owner Y wants to provide servicesfor vehicles but does not possess vehicles. Then Y may createa range-based primitive stream according to a range of areas.Any vehicle that enters the area becomes a data source of thestream automatically, and thus is able to access Y’s service.More implementation details are introduced in Section IV.Since the aggregation on an endless stream would neverreturn, the concept of window is widely accepted in traditional stream computing frameworks, such as Flink [6] andBeam [4]. It indicates the range of data to be aggregated.In Edge-Stream model, the usage of windows remains thesame. Figure 6 illustrates a stream with four data-sequences,and we focus on one of them. A fixed window divides up thesequence into fixed-width and non-overlapping time intervals.A sliding window only considers the starting point, and allowsoverlapping. For example, each window in Figure 6 (b) chunks3 seconds of data, but a new window starts every 2 seconds.The Serializer field indicates the serializing method ofdata-sequences in a stream. Given a serializer, operators areable to decode the input data and access the content properly.Common approaches include JSON, Protobuf, and Kyro.In order to clarify the usage of a window, we take thevehicle counting job as an example. Users want to figure outthe number of cars captured by each camera every hour. Theywill reuse the PlateStream generated from code 1. As shownin code listing 3, FWindow generates a new stream with anTimer Stream(a) Primitive Streamķ ĸ Ĺ(b) Sliding WindowThe following code 2 provides a simplified declaration ofa video stream and a timer stream mentioned in Figure 5.The video stream is a device-based primitive stream, whosedeclaration contains a list of devices. When the stream isin use, the system will search for those devices and inletthe data. A virtual stream declaration needs to define themethod of generating the virtual data. Once activated, theyare generated locally by the required nodes in the system. Incontrast, generated streams are defined via operators, such asthe PlateStream in code 1.A primitive stream is generated directly from endpoint physical devices. Examples include streams generated by end-point sensors, surveillance cameras, streaming media from the cloud, and so on. Therefore, the videostream in Figure 5 (a) is a primitive stream.A virtual stream introduces another type of datasource. Different from a primitive stream, it does notrely on any physical devices. Each node in an edgesystem may maintain a local replication of an identicalvirtual steam. Therefore, it is no necessary to move avirtual stream among nodes. A timer stream shown inFigure 5 (b) is a typical virtual stream, which providesthe current time every minute.A generated stream is the output generated an operator, as illustrated in Figure 5 (c). In other words, it alwaysrelies on existing streams (i.e. input of an operator). Wecall those streams as the “parent streams” of the generatedstream. For example, the license plate stream mentionedin Figure 3 is categorized as a generated stream, and thevideo stream is its parent stream.Video StreamStreamFig. 6: Two typical types of windowsAccording to the type field in metadata, streams arecategorized into three types, which are introduced as follows. ķĸĹ(a) Fixed WindowData-Sequence[ , 1:01, 1:02, 1:03.](b) Virtual Stream(c) Generated StreamFig. 5: Different types of streams Authorized licensed use limited to: University of Minnesota. Downloaded on August 11,2021 at 14:16:56 UTC from IEEE Xplore. Restrictions apply.

section implements the computation logic with C code. ThegetNext function in the first line pops one element at a timefrom the stream. The ‘%%’, ‘%{’ and ‘%}’ are punctuationto separate the sections.one-hour window from the PlateStream. Notice that FWindowis followed by a pair of parenthesis for non-stream parameters(i.e. “1h”). It is then piped to ‘VCnt’ operator to perform thefinal counting. Obvioulsy, the usage of a pipe is also similarto that in a file system.#include string #include "MyRecogLib"%%%in S video Picture, null, File %out S plate std::string, null, JSON %%%{auto inPicture S video.getNext();auto outPlate PlateRecog(inPicture);S plate.pushItem(outPlate);%}require FWindow, VCnt, Saveimport PlateStreamCountingResult FWindow("1h") PlateStream VCntStoreResult Save CountingResultListing 3: Code for the vehicle counting jobC. Operator DesignIn the Edge-Stream model, there are two types of operators,namely reshaping and computation. Reshaping operators areintroduced to define how to organize existing data-sequences,without changing the data inside. For example, a Unionoperator collects all the data-sequences from each of the inputstreams, and treat them as a new stream. The FWindowoperator in code 3 also belongs to this type. It only changeshow the system treats the data-sequences. Such concept issimilar to the string view method in C language.Computation operators, on the contrary, generate new datafrom input streams. They apply functions on each of the datasequences in the input streams. Formally, an operator withfunction f operates on a stream with three data-sequences {a,b, c} is demonstrated in Equation 1. Therefore, functions arethe pivots of designing an operator.Operatorf ({a, b, c}) {f (a), f (b), f (c)}Listing 4: Recog function implementationThe code listing 5 implements the vehicle counting operator.The input stream S plate has been segmented with a fixedwindow. The output stream S result provides the countingresult without a window. Both of them are serialized in JSON.The function intends to calculate the total number of platesin each window. The getWindow function performs in areduce-style manner. It collects data in each window andreturns a vector of elements.#include string %%%in S plate std::string, fixed, JSON %out S result int, null, JSON %%%{int counter 0;auto plates S plate.getWindow();for (plate : plates) {counter ;}S result.pushItem(counter);%}(1)Functions access data packages in each data-sequencethrough a standard set of APIs. Two of the most importantAPIs are getNext and getWindow. They correspond to twodifferent types of functions, map and reduce, respectively.The map function takes in one data-sequence from a streamand uses getNext to access each data item in the datasequence. Then, it generates a new data-sequence for theoutput stream. The reduce function involves getWindow toiterate data packages in each window of the data-sequence. Itwaits for all data in a window to accumulate in a physical nodeto start processing. They are the basic components adoptedby most of the distributed systems [9] [7] [6]. Those APIsdo not limit the data type of the operator, which supportheterogeneous data sources. The data type of the data packagesin the data-sequence is declared implicitly in the operator,demonstrated in following examples (listing 4 and 5).The code snippet in listing 4 provides a simplified implementation of the function for the Recog operator in Figure 3. Itis a map function. The code starts with an initialization sectionin C language. The following declaration part describes theinput and output streams. In the example, there is one inputstream S video consisted of Picture objects. The outputstream S plate provides std::string objects. The lastListing 5: Vehicle counting function implementationThe Edge-Stream model also supports operators to maintainper-data-sequence status for its output stream. For example,users want to know the total number of cars from now on. Theyshould replace the local variable counter with a global onefollowing a status declaration:%status counter int, 0 D. Grouping MethodTypical data processing frameworks in data-centers alwayssupport grouping methods. For example, the keyBy transformation in Flink [6] logically divides a stream into disjointpartitions. Records with the same key are ass

Edge-Stream: a Stream Processing Approach for Distributed Applications on a Hierarchical Edge-computing System Xiaoyang Wang †, Zhe Zhou , Ping Han†, Tong Meng , Guangyu Sun 1, Jidong Zhai‡ Peking University, Beijing, China, {yaoer, zhou.zhe, mengtong, gsun}@pku.edu.cn †Advanced Institute of Information Technology, Hangzhou, Zhejiang,