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

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.8489438 0.5007729      1
#> 2 0.5524538 0.8061533      2
#> 3 0.5538835 0.2362076      3
#> 4 0.9096127 0.4755122      1
#> 5 0.5888492 0.7295837      2
#> 6 0.6265946 0.2443677      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:30520 
#> 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.593 0.848 
#>  2 0.544 0.140 
#>  3 0.610 0.128 
#>  4 0.299 0.538 
#>  5 0.518 0.187 
#>  6 0.541 0.901 
#>  7 0.530 0.854 
#>  8 0.490 0.131 
#>  9 0.560 0.0906
#> 10 0.262 0.571 
#> 11 0.645 0.798 
#> 12 0.591 0.170 
#> 13 0.233 0.610 
#> 14 0.494 0.907 
#> 15 0.338 0.500 
#> 16 0.390 0.455 
#> 17 0.473 0.208 
#> 18 0.570 0.808 
#> 19 0.603 0.896 
#> 20 0.203 0.648

get_weights(clusterer)

#>  [1] 63.122068 84.135188 46.398872 84.110905 49.747729 34.429237 68.897026
#>  [8] 31.375432 39.188670 72.612374 13.765317 43.830263 29.824525 13.239980
#> [15] 42.043734  3.523916  6.523584 21.861251 12.903069  3.880440

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.461 0.119       2
#>  2 0.832 0.184       3
#>  3 0.854 0.115       3
#>  4 0.531 0.0838      2
#>  5 0.948 0.273       3
#>  6 0.735 0.437       1
#>  7 0.882 0.239       3
#>  8 0.838 0.191       3
#>  9 0.568 0.142       2
#> 10 0.870 0.206       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.