Getting Started with streamConnect

streamConnect connects data stream sources and stream mining algorithms through sockets or web services. It works with the DSD (data stream) and DSC (data stream clustering) interfaces from the stream package.

This guide shows two common workflows: reading a stream over a socket and updating a remote clusterer through a web service. For more control over the web service endpoints and deployment, see the help pages for publish_DSD_via_WebService() and publish_DSC_via_WebService().

Install and load

Install the released package from CRAN:

install.packages("streamConnect")

Load the package and select an available local port for the examples:

library(streamConnect)

#> Loading required package: stream

Read a data stream over a socket

Start a background process that publishes a stream. The publisher opens a socket server and writes data in blocks. Since the socket format does not include column names, provide them when creating the reader.

port <- httpuv::randomPort()

socket_process <- publish_DSD_via_Socket(
  DSD_Gaussians(k = 3, d = 2),
  port = port
)
socket_process

#> PROCESS 'R', running, pid 50514.

Connect a local DSD reader and request points from the stream:

socket_dsd <- DSD_ReadSocket(
  port = port,
  col.names = c("x", "y", ".class")
)

points <- get_points(socket_dsd, n = 10)
head(points)

#>            x         y .class
#> 1 0.06153633 0.1800643      3
#> 2 0.05871365 0.1982053      3
#> 3 0.60970641 0.3918385      2
#> 4 0.55826419 0.3602819      2
#> 5 0.06230688 0.1740417      3
#> 6 0.03009396 0.2419788      3

Close the reader and stop the publisher when finished:

close_stream(socket_dsd)
if (socket_process$is_alive()) socket_process$kill()

#> [1] TRUE

Update a remote clusterer via a web service

Web services also support request and response operations. Start a clusterer in a background process, then create a local DSC interface for it:

web_port <- httpuv::randomPort()

clusterer_process <- publish_DSC_via_WebService(
  "DSC_DBSTREAM(r = .05)",
  port = web_port
)

clusterer <- DSC_WebService(paste0("http://localhost:", web_port))
clusterer

#> Web Service Data Stream Clusterer: DBSTREAM
#> Served from: http://localhost:6876 
#> Class: DSC_WebService, DSC_R, DSC 
#> Number of micro-clusters: 0 
#> Number of macro-clusters: 0

Send a batch of points to the remote clusterer and retrieve its centers and weights through the local interface:

stream <- DSD_Gaussians(k = 3, d = 2, noise = 0.05)
update(clusterer, stream, n = 500)

get_centers(clusterer)

#> # A tibble: 23 × 2
#>        X1      X2
#>     <dbl>   <dbl>
#>  1 0.520  0.0902 
#>  2 0.461  0.00199
#>  3 0.483  0.0452 
#>  4 0.320  0.775  
#>  5 0.141  0.411  
#>  6 0.107  0.380  
#>  7 0.312  0.825  
#>  8 0.528  0.0365 
#>  9 0.0515 0.331  
#> 10 0.370  0.829  
#> # ℹ 13 more rows

get_weights(clusterer)

#>  [1] 58.075120 41.522420 75.577231 81.433853 60.090369 78.808096 47.375853
#>  [8] 57.390820 44.246291 37.918858 55.624115  4.155875 14.468104  4.117236
#> [15]  3.332356 15.104319 10.438558  4.535522  9.450619 12.676356 16.438439
#> [22] 30.433002 17.653344

Stop the service process when the work is complete:

if (clusterer_process$is_alive()) clusterer_process$kill()

#> [1] TRUE

Publish a stream through a web service

To expose a data source over HTTP, publish a DSD and read it with DSD_ReadWebService(). The service returns CSV by default; JSON and RDS serialization are also available through the serializer argument.

port <- httpuv::randomPort()

source_process <- publish_DSD_via_WebService(
  "DSD_Gaussians(k = 3, d = 2)",
  port = port
)

remote_stream <- DSD_ReadWebService(paste0("http://localhost:", port))
get_points(remote_stream, n = 10)

#> # A tibble: 10 × 3
#>        X1     X2 .class
#>     <dbl>  <dbl>  <dbl>
#>  1 0.741  0.0196      1
#>  2 0.0837 0.615       3
#>  3 0.230  0.442       2
#>  4 0.707  0.0735      1
#>  5 0.143  0.677       3
#>  6 0.190  0.633       3
#>  7 0.221  0.429       2
#>  8 0.737  0.0794      1
#>  9 0.239  0.326       2
#> 10 0.186  0.469       2

if (source_process$is_alive()) source_process$kill()

#> [1] TRUE

Next steps

Use publish_DSD_via_Socket() and DSD_ReadSocket() for continuous point-to-point stream transfer.
Use publish_DSD_via_WebService() and DSD_ReadWebService() to serve stream data over HTTP.
Use publish_DSC_via_WebService() and DSC_WebService() to update and inspect a remote clusterer.
Set serve = FALSE to write a plumber task file for deployment with a hosting environment of your choice.