Getting Started with streamConnect

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 5416.

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.3909010 0.5082749      2
#> 2 0.3787903 0.4833002      2
#> 3 0.7908420 0.8276591      1
#> 4 0.3921697 0.4857952      2
#> 5 0.7433751 0.7931640      1
#> 6 0.3656962 0.5809681      2

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:8379 
#> 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: 21 × 2
#>        X1    X2
#>     <dbl> <dbl>
#>  1 0.0434 0.596
#>  2 1.00   0.760
#>  3 0.0950 0.590
#>  4 0.395  0.229
#>  5 0.966  0.734
#>  6 0.0588 0.549
#>  7 0.419  0.159
#>  8 0.367  0.187
#>  9 0.113  0.525
#> 10 0.462  0.129
#> # ℹ 11 more rows

get_weights(clusterer)

#>  [1] 59.704327 64.380509 51.963242 31.583712 95.004162 43.812891 74.873811
#>  [8] 44.308259 26.732697 44.127368  4.294116 27.198350 23.685512  9.800871
#> [15] 19.066719 39.109756 17.120931 19.603789  3.720082  4.769581  4.722892

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.251 0.182      3
#>  2 0.286 0.180      3
#>  3 0.281 0.182      3
#>  4 0.530 0.215      1
#>  5 0.183 0.128      3
#>  6 0.211 0.187      3
#>  7 0.234 0.177      3
#>  8 0.913 0.417      2
#>  9 0.607 0.285      1
#> 10 0.150 0.172      3

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.

See the package reference for the full function documentation.