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

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.84251016 0.6776340      1
#> 2 0.09981386 0.3078319      3
#> 3 0.83820639 0.7009123      1
#> 4 0.09701102 0.3874605      3
#> 5 0.80188514 0.6306109      1
#> 6 0.85946761 0.7023793      1

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:48387 
#> 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: 22 × 2
#>       X1    X2
#>    <dbl> <dbl>
#>  1 0.978 0.818
#>  2 0.619 0.753
#>  3 0.167 0.855
#>  4 0.218 0.872
#>  5 0.630 0.698
#>  6 0.950 0.858
#>  7 0.570 0.690
#>  8 0.122 0.830
#>  9 0.906 0.840
#> 10 0.259 0.928
#> # ℹ 12 more rows

get_weights(clusterer)

#>  [1] 71.747882 75.599247 72.578039 62.925623 75.323826 57.288661 48.990913
#>  [8] 40.231167 55.150923  5.905853 38.519308  8.849252 60.335906  3.282414
#> [15] 37.085182 11.498260 11.247490 15.912668  8.121847 20.168858  8.967682
#> [22] 10.342084

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.0787  0.612      2
#>  2 0.166   0.696      2
#>  3 0.00172 0.517      2
#>  4 0.889   0.422      1
#>  5 0.470   0.997      3
#>  6 0.553   0.949      3
#>  7 0.0943  0.590      2
#>  8 0.534   0.910      3
#>  9 0.191   0.713      2
#> 10 0.101   0.626      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.

See the package reference for the full function documentation.