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

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.7803439 0.8030971      2
#> 2 0.8078972 0.7784756      2
#> 3 0.9323115 0.1387254      3
#> 4 0.9429125 0.2134097      3
#> 5 0.8960368 0.6872709      2
#> 6 0.9119127 0.2279724      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:48798 
#> 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: 27 × 2
#>          X1    X2
#>       <dbl> <dbl>
#>  1  0.587   0.966
#>  2 -0.00367 0.829
#>  3  0.352   0.520
#>  4  0.535   0.991
#>  5 -0.0700  0.744
#>  6  0.432   0.917
#>  7  0.393   0.485
#>  8  0.504   0.928
#>  9  0.475   0.894
#> 10  0.550   0.919
#> # ℹ 17 more rows

get_weights(clusterer)

#>  [1]  6.643676 60.844286 75.415331  5.944662 19.747763 44.067063 50.334214
#>  [8] 41.706901 55.481345 20.793782 15.402825 54.290626 50.327502 25.324523
#> [15] 58.350708 17.694832  6.132422 33.895684 11.210299 33.134705 14.862854
#> [22]  8.674450 15.045803 11.258698 20.688059  6.683722  4.727086

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.134 0.117      3
#>  2 0.192 0.135      3
#>  3 0.176 0.165      3
#>  4 0.177 0.147      3
#>  5 0.577 0.176      2
#>  6 0.380 0.199      1
#>  7 0.385 0.211      1
#>  8 0.512 0.167      2
#>  9 0.604 0.200      2
#> 10 0.149 0.115      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.