Add event stream batching (#62)

* Add support for batching events at stream level.

* Add example with both batching and multitask operators.

* Add batching timeout.

* Add HTTP and slack combined batching examples.

Author: doru1004

examples/batching/batching.py

@@ -0,0 +1,75 @@
+#
+# Copyright IBM Corporation 2021
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import ray
+import rayvens
+import sys
+
+# Send message to Slack sink using a stream which batches events.
+
+# Command line arguments and validation:
+if len(sys.argv) < 4:
+    print(f'usage: {sys.argv[0]} <slack_channel> <slack_webhook> <run_mode>')
+    sys.exit(1)
+slack_channel = sys.argv[1]
+slack_webhook = sys.argv[2]
+run_mode = sys.argv[3]
+if run_mode not in ['local', 'mixed', 'operator']:
+    raise RuntimeError(f'Invalid run mode provided: {run_mode}')
+
+# Initialize ray either on the cluster or locally otherwise.
+if run_mode == 'operator':
+    ray.init(address='auto')
+else:
+    ray.init()
+
+# Start rayvens in operator mode.
+rayvens.init(mode=run_mode)
+
+# Create stream.
+stream = rayvens.Stream('slack', batch_size=2)
+
+
+# Operator task:
+@ray.remote
+def batching_operator(incoming_events):
+    print(incoming_events)
+    return " ".join(incoming_events)
+
+
+# Event sink config.
+sink_config = dict(kind='slack-sink',
+                   route='/toslack',
+                   channel=slack_channel,
+                   webhook_url=slack_webhook)
+
+# Add sink to stream.
+sink = stream.add_sink(sink_config)
+
+# Add multi-task operator to stream.
+stream.add_operator(batching_operator)
+
+# Sends messages to all sinks attached to this stream.
+stream << "Hello"
+stream << "World"
+stream << "Hello"
+stream << "Mars"
+stream << "Hello"
+stream << "Jupiter"
+
+# Disconnect any sources or sinks attached to the stream 2 seconds after
+# the stream is idle (i.e. no events were propagated by the stream).
+stream.disconnect_all(after_idle_for=2)

examples/batching/batching_source.py

@@ -0,0 +1,84 @@
+#
+# Copyright IBM Corporation 2021
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import ray
+import rayvens
+import sys
+import json
+
+# Send message to Slack sink using a stream which batches events received
+# from an HTTP source.
+
+# Command line arguments and validation:
+if len(sys.argv) < 4:
+    print(f'usage: {sys.argv[0]} <slack_channel> <slack_webhook> <run_mode>')
+    sys.exit(1)
+slack_channel = sys.argv[1]
+slack_webhook = sys.argv[2]
+run_mode = sys.argv[3]
+if run_mode not in ['local', 'mixed', 'operator']:
+    raise RuntimeError(f'Invalid run mode provided: {run_mode}')
+
+# Initialize ray either on the cluster or locally otherwise.
+if run_mode == 'operator':
+    ray.init(address='auto')
+else:
+    ray.init()
+
+# Start rayvens in operator mode.
+rayvens.init(mode=run_mode)
+
+# Create stream.
+stream = rayvens.Stream('slack', batch_size=2)
+
+
+# Operator task:
+@ray.remote
+def batching_operator(incoming_events):
+    quotes = []
+    for event in incoming_events:
+        payload = json.loads(event)  # parse event string to json
+        quotes.append(
+            str(payload['quoteResponse']['result'][0]['regularMarketPrice']))
+    return " ".join(quotes)
+
+
+# Event sink config.
+sink_config = dict(kind='slack-sink',
+                   route='/toslack',
+                   channel=slack_channel,
+                   webhook_url=slack_webhook)
+
+# Add sink to stream.
+sink = stream.add_sink(sink_config)
+
+# Add multi-task operator to stream.
+stream.add_operator(batching_operator)
+
+# Event source config.
+source_config = dict(
+    kind='http-source',
+    name='source-1',
+    url='https://query1.finance.yahoo.com/v7/finance/quote?symbols=AAPL',
+    route='/from-http',
+    period=3000)
+
+# Attach source to stream.
+source = stream.add_source(source_config)
+
+# Disconnect any sources or sinks attached to the stream 2 seconds after
+# the stream is idle (i.e. no events were propagated by the stream).
+stream.disconnect_all(after=10)

examples/batching/batching_with_multitask.py

@@ -0,0 +1,82 @@
+#
+# Copyright IBM Corporation 2021
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import ray
+import rayvens
+import sys
+
+# Send message to Slack sink using a stream which batches events and a
+# multi-task operator.
+
+# Command line arguments and validation:
+if len(sys.argv) < 4:
+    print(f'usage: {sys.argv[0]} <slack_channel> <slack_webhook> <run_mode>')
+    sys.exit(1)
+slack_channel = sys.argv[1]
+slack_webhook = sys.argv[2]
+run_mode = sys.argv[3]
+if run_mode not in ['local', 'mixed', 'operator']:
+    raise RuntimeError(f'Invalid run mode provided: {run_mode}')
+
+# Initialize ray either on the cluster or locally otherwise.
+if run_mode == 'operator':
+    ray.init(address='auto')
+else:
+    ray.init()
+
+# Start rayvens in operator mode.
+rayvens.init(mode=run_mode)
+
+# Create stream.
+stream = rayvens.Stream('slack', batch_size=6)
+
+
+# Operator sub-task:
+@ray.remote
+def sub_task(context, sub_event):
+    context.publish(sub_event)
+
+
+# Operator task:
+@ray.remote
+def batching_multi_task_operator(context, incoming_events):
+    for sub_event in incoming_events:
+        sub_task.remote(context, sub_event)
+
+
+# Event sink config.
+sink_config = dict(kind='slack-sink',
+                   route='/toslack',
+                   channel=slack_channel,
+                   webhook_url=slack_webhook)
+
+# Add sink to stream.
+sink = stream.add_sink(sink_config)
+
+# Add multi-task operator to stream.
+stream.add_multitask_operator(batching_multi_task_operator)
+
+# Sends messages to all sinks attached to this stream.
+stream << "Hello"
+stream << "World"
+stream << "Hello"
+stream << "Mars"
+stream << "Hello"
+stream << "Jupiter"
+
+# Disconnect any sources or sinks attached to the stream 2 seconds after
+# the stream is idle (i.e. no events were propagated by the stream).
+stream.disconnect_all(after_idle_for=2)

rayvens/api.py

@@ -25,6 +25,7 @@
 from rayvens.core.ray_serve import start as start_operator_ray_serve
 from rayvens.core.name import name_source, name_sink
 from rayvens.core.verify import verify_do
+from queue import Queue
 
 
 class Stream:
@@ -33,13 +34,14 @@ def __init__(self,
                  actor_options=None,
                  operator=None,
                  source_config=None,
-                 sink_config=None):
+                 sink_config=None,
+                 batch_size=None):
         if _global_camel is None:
             raise RuntimeError(
                 "Rayvens has not been started. Start with 'rayvens.init()'.")
         self.name = name
         self.actor = StreamActor.options(actor_options).remote(
-            name, operator=operator)
+            name, operator=operator, batch_size=batch_size)
         if sink_config is not None:
             self.add_sink(sink_config)
         if source_config is not None:
@@ -53,11 +55,12 @@ def append(self, data):
         self.actor.append.remote(data)
         return self
 
-    def add_operator(self, operator):
-        return ray.get(self.actor.add_operator.remote(operator))
+    def add_operator(self, operator, batch_size=None):
+        return ray.get(self.actor.add_operator.remote(operator, batch_size))
 
-    def add_multi_operator(self, operator):
-        return ray.get(self.actor.add_multi_operator.remote(operator))
+    def add_multitask_operator(self, operator, batch_size=None):
+        return ray.get(
+            self.actor.add_multitask_operator.remote(operator, batch_size))
 
     def add_source(self, source_config):
         return ray.get(self.actor.add_source.remote(self, source_config))
@@ -116,13 +119,14 @@ def _idle_time(self):
 
 
 class StreamContext:
-    def __init__(self):
+    def __init__(self, stream_batch_size):
         self.sink_restrictions = {}
         self.subscribers = {}
         self.latest_sent_event_timestamp = None
         self.event_counter = 0
         self.limit_subscribers = False
         self.is_multi_operator = False
+        self.stream_batch_size = stream_batch_size
 
     def publish(self, data):
         if data is not None:
@@ -148,12 +152,15 @@ def _accepts_data_type(self, data, type_restrictions):
 
 @ray.remote(num_cpus=0)
 class StreamActor:
-    def __init__(self, name, operator=None):
+    def __init__(self, name, operator=None, batch_size=None):
         self.name = name
         self._operator = operator
         self._sources = {}
         self._sinks = {}
-        self.context = StreamContext()
+        self.context = StreamContext(batch_size)
+        self.queue = None
+        if self.context.stream_batch_size is not None:
+            self.queue = Queue(maxsize=self.context.stream_batch_size)
 
     def send_to(self, subscriber, name=None):
         if self.context.limit_subscribers:
@@ -168,17 +175,26 @@ def send_to(self, subscriber, name=None):
     def append(self, data):
         if data is None:
             return
+        if self.queue is not None:
+            self.queue.put_nowait(data)
+            if not self.queue.full():
+                return
+            data = list(self.queue.queue)
+            with self.queue.mutex:
+                self.queue.queue.clear()
         if self._operator is not None:
             data = _eval(self.context, self._operator, data)
         self.context.publish(data)
         self.context.latest_sent_event_timestamp = time.time()
         self.context.event_counter += 1
 
-    def add_operator(self, operator):
+    def add_operator(self, operator, batch_size):
         self._operator = operator
+        self.context.batch_size = batch_size
 
-    def add_multitask_operator(self, operator):
+    def add_multitask_operator(self, operator, batch_size):
         self._operator = operator
+        self.context.batch_size = batch_size
         self.context.is_multi_operator = True
 
     def add_source(self, stream, source_config):
@@ -229,13 +245,29 @@ def disconnect_sink(self, sink_name):
     def disconnect_all(self, stream_drain_timeout):
         for source_name in dict(self._sources):
             self.disconnect_source(source_name)
+        if self.queue is not None:
+            self.flush_batch()
         time.sleep(stream_drain_timeout)
         for sink_name in dict(self._sinks):
             self.disconnect_sink(sink_name)
 
     def event_count(self):
         return self.context.event_counter
 
+    def flush_batch(self):
+        data = None
+        if self.queue is not None:
+            data = list(self.queue.queue)
+            with self.queue.mutex:
+                self.queue.queue.clear()
+        if data is None:
+            return
+        if self._operator is not None:
+            data = _eval(self.context, self._operator, data)
+        self.context.publish(data)
+        self.context.latest_sent_event_timestamp = time.time()
+        self.context.event_counter += 1
+
     def _meta(self, action, *args, **kwargs):
         return verify_do(self, _global_camel, action, *args, **kwargs)