Ray Streaming
=============

Ray Streaming is a streaming data processing framework built on ray. It will be helpful for you to build jobs dealing with real-time data.

Key Features
------------


#.
**Cross Language**. Based on Ray's multi-language actor, Ray Streaming can also run in multiple
languages(only Python and Java is supported currently) with high efficiency. You can implement your
operator in different languages and run them in one job.

#.
**Single Node Failover**. We designed a special failover mechanism that only needs to rollback the
failed node it's own, in most cases, to recover the job. This will be a huge benefit if your job is
sensitive about failure recovery time. In other frameworks like Flink, instead, the entire job should
be restarted once a node has failure.

Examples
--------

Python
^^^^^^

.. code-block:: Python

import ray
from ray.streaming import StreamingContext

ctx = StreamingContext.Builder() \
.build()
ctx.read_text_file(__file__) \
.set_parallelism(1) \
.flat_map(lambda x: x.split()) \
.map(lambda x: (x, 1)) \
.key_by(lambda x: x[0]) \
.reduce(lambda old_value, new_value:
(old_value[0], old_value[1] + new_value[1])) \
.filter(lambda x: "ray" not in x) \
.sink(lambda x: print("result", x))
ctx.submit("word_count")

Java
^^^^

.. code-block:: Java

StreamingContext context = StreamingContext.buildContext();
List text = Collections.singletonList("hello world");
DataStreamSource.fromCollection(context, text)
.flatMap((FlatMapFunction) (value, collector) -> {
String[] records = value.split(" ");
for (String record : records) {
collector.collect(new WordAndCount(record, 1));
}
})
.filter(pair -> !pair.word.contains("world"))
.keyBy(pair -> pair.word)
.reduce((oldValue, newValue) ->
new WordAndCount(oldValue.word, oldValue.count + newValue.count))
.sink(result -> System.out.println("sink result=" + result));
context.execute("testWordCount");

Use Java Operators in Python
^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.. code-block:: Python

import ray
from ray.streaming import StreamingContext

ctx = StreamingContext.Builder().build()
ctx.from_values("a", "b", "c") \
.as_java_stream() \
.map("io.ray.streaming.runtime.demo.HybridStreamTest$Mapper1") \
.filter("io.ray.streaming.runtime.demo.HybridStreamTest$Filter1") \
.as_python_stream() \
.sink(lambda x: print("result", x))
ctx.submit("HybridStreamTest")

Use Python Operators in Java
^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.. code-block:: Java

StreamingContext context = StreamingContext.buildContext();
DataStreamSource streamSource =
DataStreamSource.fromCollection(context, Arrays.asList("a", "b", "c"));
streamSource
.map(x -> x + x)
.asPythonStream()
.map("ray.streaming.tests.test_hybrid_stream", "map_func1")
.filter("ray.streaming.tests.test_hybrid_stream", "filter_func1")
.asJavaStream()
.sink(value -> System.out.println("HybridStream sink=" + value));
context.execute("HybridStreamTestJob");

Installation
------------

Python
^^^^^^

Ray Streaming is packaged together with Ray, install Ray with: ``pip install ray``\ ,
this wheel contains all dependencies your need to run Python streaming, including Java operators supporting.

Java
^^^^

Import Ray Streaming using maven:

.. code-block:: xml


ray-api
io.ray
1.0.1


ray-runtime
io.ray
1.0.1


streaming-api
io.ray
1.0.1


streaming-runtime
io.ray
1.0.1


Internal Design
---------------

Overall Architecture
^^^^^^^^^^^^^^^^^^^^


.. image:: assets/architecture.jpg
:target: assets/architecture.jpg
:alt: architecture


Ray Streaming is built on Ray. We use Ray's actor to run everything, and use Ray's direct call for communication.

There are two main types of actor: job master and job worker.

When you execute ``context.submit()`` in your driver, we'll first create a job master, then job master will create all job workers needed to run your operator. Then job master will be responsible to coordinate all workers, including checkpoint, failover, etc.

Check `Ray Streaming Proposal `_
to get more detailed information about the overall design.

Fault Tolerance Mechanism
^^^^^^^^^^^^^^^^^^^^^^^^^

As mentioned above, different from other frameworks, We designed a special failover mechanism that only needs to rollback the failed node it's own, in most cases, to recover the job. The main idea to achieve this feature is saving messages for each node, and replay them from upstream when node has failure.

Check `Fault Tolerance Proposal `_
for more detailed information about our fault tolerance mechanism.

Development Guides
------------------


#.
Build streaming java


* build ray

* ``bazel build //java:gen_maven_deps``
* ``cd java && mvn clean install -Dmaven.test.skip=true && cd ..``

* build streaming

* ``bazel build //streaming/java:gen_maven_deps``
* ``mvn clean install -Dmaven.test.skip=true``

#.
Build ray python will build ray streaming python.

#.
Run examples

.. code-block:: bash

# c++ test
cd streaming/ && bazel test ...
sh src/test/run_streaming_queue_test.sh
cd ..

# python test
pushd python/ray/streaming/
pushd examples
python simple.py --input-file toy.txt
popd
pushd tests
pytest .
popd
popd

# java test
cd streaming/java/streaming-runtime
mvn test


More Information
----------------


* `Ray Streaming implementation plan `_
* `Fault Tolerance Proposal `_
* `Data Transfer Proposal `_
* `Ray Streaming Proposal `_
* `Open Source Plan `_

Getting Involved
----------------

- `Community Slack`_: Join our Slack workspace.
- `GitHub Discussions`_: For discussions about development, questions about usage, and feature requests.
- `GitHub Issues`_: For reporting bugs.

.. _`GitHub Discussions`: https://github.com/ray-project/ray/discussions
.. _`GitHub Issues`: https://github.com/ray-project/ray/issues
.. _`Community Slack`: https://forms.gle/9TSdDYUgxYs8SA9e8