stream: Working With Data Streams using Connections and Web Services

Introduction

Data streams are often processed in a distributed manner using multiple machines or multiple processes. For example, a data stream may be produced by a sensor attached to a remote machine or multiple clustering algorithms run in parallel using several R processes. Another application is to connect to other software components in a stream mining pipeline.

First, we show how socket connections together with the package stream can be used to connect multiple processes or machines.

Then we give examples of how package streamConnect makes connecting stream mining components more convenient by providing an interface to connect stream processing using sockets or web services. While sockets are only used to connect data steam generating processes, web services are more versatile and can also be used to create data stream clustering processes as a service.

The final section of this paper shows how to deploy the server/web service.

Using Sockets Directly to Publish a Data Stream

The functions write_stream() and the class DSD_ReadStream provided in package stream can be used for communicate via connections (files, sockets, URLs, etc.). In the first example, we manually set up the connection. The example is useful to understand how sockets work especially for users interested in implementing their own components using other programming languages or connecting with other data stream software.

A more convenient way to do this using package streamConnect is described later in this paper.

For we find an available port.

port <- httpuv::randomPort()
port
## [1] 42515

Server: Serving a Data Stream

The server serves data from a data stream. We use library callr to create a separate R process that serves a data stream creating 10 points every second using a socket connection, but you can also put the code in function r_bg() in a file called server.R and run (potentially on a different machine) it with R CMD BATCH server.R from the command line.

library(stream)
library(callr)
## 
## Attaching package: 'callr'
## The following object is masked from 'package:rmarkdown':
## 
##     run
rp1 <- r_bg(function(port) {
  library(stream)
  stream <- DSD_Gaussians(k = 3, d = 3)
  blocksize <- 10

  con <- socketConnection(port = port, server = TRUE)
  
  while (TRUE) {
    write_stream(stream, con, n = blocksize, close = FALSE)
    Sys.sleep(1)
  }
  
  close(con)
}, 
  args = list(port = port))

rp1
## PROCESS 'R', running, pid 3923.

Client: Reading from the Stream

The client consumes the data stream. We open the connection which starts the data generating process. Note that streamConnect is not used here. For convenience, we only use the helper retry() defined in streamConnect to make sure the server connections are established.

con <- streamConnect::retry(socketConnection(port = port, open = 'r'))
con
## A connection with                               
## description "->localhost:42515"
## class       "sockconn"         
## mode        "r"                
## text        "text"             
## opened      "opened"           
## can read    "yes"              
## can write   "yes"
dsd <- streamConnect::retry(DSD_ReadStream(con))

We poll all available data (n = -1) several times. The first request should yield 10 points, the second none and the third request should yield 20 points (2 seconds).

get_points(dsd, n= -1)
##            V1        V2        V3
## 1  0.84194267 0.6886325 0.1870898
## 2  0.88315996 0.6847950 0.2123483
## 3  0.82181290 0.6317726 0.1192483
## 4  0.80284620 0.6237352 0.1490156
## 5  0.05361430 0.3702645 0.6478312
## 6  0.57370265 0.6866024 0.4345926
## 7  0.87956303 0.6608399 0.2136718
## 8  0.51307536 0.5915974 0.4087341
## 9  0.82302634 0.6392184 0.1825586
## 10 0.85250990 0.6453043 0.1951930
## 11 0.57947003 0.6510139 0.4835289
## 12 0.84460818 0.6488912 0.2021404
## 13 0.56048964 0.6873854 0.4682796
## 14 0.78487010 0.6284412 0.1676875
## 15 0.07617283 0.4049309 0.6046798
## 16 0.84371925 0.6213517 0.1555107
## 17 0.83008433 0.6153023 0.1505546
## 18 0.82015986 0.6414969 0.1692893
## 19 0.84735763 0.6606163 0.1709363
## 20 0.08406526 0.3569909 0.6355549
get_points(dsd, n= -1)
## [1] V1 V2 V3
## <0 rows> (or 0-length row.names)
Sys.sleep(2)
get_points(dsd, n= -1)
##             V1        V2        V3
## 1  0.009093626 0.3998528 0.6178337
## 2  0.089342988 0.3221792 0.6627430
## 3  0.830645507 0.6392772 0.1636347
## 4  0.874150772 0.6340835 0.1885534
## 5  0.001841278 0.3499958 0.6168476
## 6  0.015086312 0.4034213 0.6709696
## 7  0.824160914 0.6607016 0.1897861
## 8  0.080453105 0.4206395 0.6433582
## 9  0.535294181 0.7616758 0.5296346
## 10 0.809581364 0.6027896 0.1029431
## 11 0.817079894 0.6552704 0.1642743
## 12 0.075504060 0.4061076 0.6515615
## 13 0.591435796 0.7209192 0.5045354
## 14 0.048753489 0.3169766 0.6374953
## 15 0.584888971 0.6736746 0.4551359
## 16 0.098884340 0.3809745 0.6171123
## 17 0.873513372 0.6722242 0.2157143
## 18 0.064791105 0.3794545 0.6271446
## 19 0.626932505 0.6545402 0.4335186
## 20 0.810322892 0.6283721 0.1563945
close(con)

Server: Stoping the Server Process

Here we stop the callr process. Note that the socket connection is still active and will serve the data in the connection buffer as long as the reading process keeps the connection open.

rp1$kill()
## [1] TRUE

streamConnect Sockets

streamConnect provides a more convenient way to set up a connection using sockets. publish_DSD_via_Socket() creates a socket broadcasting the data stream and DSD_ReadSocket creates a DSD object reading from that socket.

We will use an available port.

port <- httpuv::randomPort()
port
## [1] 4191

Server: Publish Data

We create a DSD process sending data to the port.

library(streamConnect)

rp1 <- DSD_Gaussians(k = 3, d = 3) %>% publish_DSD_via_Socket(port = port)
rp1
## PROCESS 'R', running, pid 3978.

Client: Connect to the Data Stream

Next, we create a DSD that connects to the socket. DSD_ReadSocket() already performs internally retries

library(streamConnect)

dsd <- DSD_ReadSocket(port = port, col.names = c("x", "y", "z", ".class"))
dsd
## Data Stream from Connection (d = 3, k = NA) 
## Class: DSD_ReadStream, DSD_R, DSD 
## connection: ->localhost:4191 (opened)
get_points(dsd, n = 10)
##            x            y         z .class
## 1  0.3059133  0.786239288 0.7410889      1
## 2  0.7819349 -0.002441458 0.7713916      2
## 3  0.5018785  0.906690431 0.2642709      3
## 4  0.6448571  0.047007527 0.7361342      2
## 5  0.6394739  0.073368809 0.7683808      2
## 6  0.5869878  0.090355251 0.6938564      2
## 7  0.3231510  0.712829254 0.6628383      1
## 8  0.5025411  0.865051780 0.2680854      3
## 9  0.3177218  0.770013661 0.7005005      1
## 10 0.5788185  0.884408462 0.2670759      3
plot(dsd)

close_stream(dsd)

Server: Stoping the Server Process

Closing the stream on the client also closes the connection which may already kill the serving process.

if (rp1$is_alive()) rp1$kill()

streamConnect Web Services

Web services are more versatile, they can be used to deploy data stream sources using publish_DSD_via_WebService()/DSD_ReadWebservice or data stream tasks using publish_DSC_via_WebService()/DSC_WebService. Here we only show how to deploy a clusterer, but a DSD can be published in a similar manner. Larger workflows can be created using DST_Runner from stream.

streamConnect uses the package plumber to manage web services. The data is transmitted in serialized form. The default serialization format it csv (comma separated values). Other formats are json and rds (see plumber::serializer_csv).

We will use an available port.

port <- httpuv::randomPort()
port
## [1] 17661

Server: Create a Web Service

Creating a clustering web service process listening for data on the port.

library(streamConnect)

rp1 <- publish_DSC_via_WebService("DSC_DBSTREAM(r = .05)", port = port)
rp1
## PROCESS 'R', running, pid 4033.

Client: Connect to the Web Service

Connect to the web service with a local DSC interface.

library(streamConnect)

dsc <- DSC_WebService(paste0("http://localhost", ":", port), 
                      verbose = TRUE, config = httr::verbose(info = TRUE))
## Connecting to DSC Web service at http://localhost:17661
## Success
dsc
## Web Service Data Stream Clusterer: DBSTREAM
## Served from: http://localhost:17661 
## Class: DSC_WebService, DSC_R, DSC 
## Number of micro-clusters: 0 
## Number of macro-clusters: 0

Note that the verbose output can help with debugging connection issues.

Cluster some data.

dsd <- DSD_Gaussians(k = 3, d = 2, noise = 0.05)

update(dsc, dsd, 500)
dsc
## Web Service Data Stream Clusterer: DBSTREAM
## Served from: http://localhost:17661 
## Class: DSC_WebService, DSC_R, DSC 
## Number of micro-clusters: 19 
## Number of macro-clusters: 3
get_centers(dsc)
## # A tibble: 19 × 2
##        X1     X2
##     <dbl>  <dbl>
##  1 0.0988 0.480 
##  2 0.122  0.382 
##  3 0.522  0.118 
##  4 0.193  0.372 
##  5 0.323  0.732 
##  6 0.503  0.0742
##  7 0.282  0.704 
##  8 0.241  0.381 
##  9 0.0680 0.409 
## 10 0.375  0.741 
## 11 0.560  0.154 
## 12 0.158  0.409 
## 13 0.122  0.440 
## 14 0.248  0.653 
## 15 0.434  0.0464
## 16 0.604  0.195 
## 17 0.0434 0.469 
## 18 0.323  0.675 
## 19 0.336  0.783
get_weights(dsc)
##  [1] 17.332313 73.835209 71.041960 23.604311 70.188219 46.346123 68.950410
##  [8]  4.130247 52.822135 39.839400 48.185709 61.349648 60.449194 27.971120
## [15]  6.661200  8.981933  6.358154 43.725272 16.285698
plot(dsc)

Server: Stop the Web Service

Kill the web service process.

rp1$kill()
## [1] TRUE

Deploying the Server/Web Service

Web services and the socket-based server can be easily deployed to any server or cloud system including containers. Make sure R and the package streamConnect and all dependencies are installed. Create a short R script to start the server/service and deploy it.

library(streamConnect)
port = 8001

publish_DSC_via_WebService("DSC_DBSTREAM(r = .05)", port = port, 
                           background = FALSE)

Web services can also be deployed using a plumber task file. The following call does not create a server, but returns the name of the task file.

publish_DSC_via_WebService("DSC_DBSTREAM(r = .05)", serve = FALSE)

Open the file in R studio to deploy it or read the plumber Hosting vignette.