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

We will use the same free server port throughout this vignette. Some systems do not allow the use of local ports, then the code in this vignette will not be executed.

run_code <- TRUE
tryCatch({
  port <- httpuv::randomPort()

  }, warning = function(w) {
    message("Ports unavailable: ", w)
    run_code <<- FALSE
  }, error = function(e) {
    message("Ports unavailable: ", e)
    run_code <<- FALSE
})

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.

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

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

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.6800171 0.8880206      1
#> 2 0.2142635 0.1373371      2
#> 3 0.3880251 0.7212369      3
#> 4 0.6784146 0.9536051      1
#> 5 0.5979322 0.9152445      1
#> 6 0.4995153 0.7062725      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:

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

clusterer <- DSC_WebService(paste0("http://127.0.0.1:", port))
clusterer

#> Web Service Data Stream Clusterer: DBSTREAM
#> Served from: http://127.0.0.1:34986 
#> 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: 20 × 2
#>         X1    X2
#>      <dbl> <dbl>
#>  1 0.0621  0.192
#>  2 0.931   0.633
#>  3 0.0237  0.222
#>  4 0.312   0.765
#>  5 1.00    0.692
#>  6 0.877   0.607
#>  7 0.390   0.794
#>  8 0.241   0.691
#>  9 0.353   0.739
#> 10 0.289   0.717
#> 11 0.00913 0.269
#> 12 0.109   0.155
#> 13 0.426   0.824
#> 14 0.0780  0.238
#> 15 0.892   0.666
#> 16 0.919   0.587
#> 17 0.984   0.645
#> 18 0.342   0.808
#> 19 0.950   0.682
#> 20 0.829   0.588

get_weights(clusterer)

#>  [1] 78.854674 80.542163 63.682946 75.065103  5.895465 69.291220 50.678596
#>  [8] 10.433820 49.889270 47.564686 18.534335 26.291038  5.255817 39.305810
#> [15] 37.867081 42.786485 21.799254 33.222428 18.026878  3.836443

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.

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

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

#> # A tibble: 10 × 3
#>       X1     X2 .class
#>    <dbl>  <dbl>  <dbl>
#>  1 0.655 0.0439      2
#>  2 0.384 0.768       1
#>  3 0.397 0.722       1
#>  4 0.619 0.0547      2
#>  5 0.330 0.823       1
#>  6 0.396 0.676       1
#>  7 0.376 0.844       1
#>  8 0.667 0.0480      2
#>  9 0.446 0.713       1
#> 10 0.605 0.0406      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.