0
# Ray Streaming
1
2
A distributed streaming processing framework built on Ray that provides fault-tolerant, scalable stream processing with Python API and cross-language support.
3
4
## Package Information
5
6
- **Package Name**: ray (streaming module)
7
- **Package Type**: PyPI
8
- **Language**: Python
9
- **Installation**: `pip install ray`
10
11
## Overview
12
13
Ray Streaming provides a distributed stream processing framework built on Ray's actor model. It enables fault-tolerant stream processing with automatic checkpointing, resource management, and cross-language integration between Python and Java workloads.
14
15
**Key Features:**
16
- Distributed stream processing with fault tolerance
17
- Single-node failover mechanism for fast recovery
18
- Cross-language support (Python and Java operators)
19
- Built-in checkpointing and state management
20
- Integration with Ray's distributed computing capabilities
21
22
## Core Imports
23
24
```python
25
from ray.streaming import StreamingContext
26
from ray.streaming.datastream import DataStream, StreamSource
27
from ray.streaming.function import SourceFunction, CollectionSourceFunction
28
```
29
30
## Basic Usage
31
32
### Simple Word Count Example
33
34
```python
35
import ray
36
from ray.streaming import StreamingContext
37
38
# Initialize Ray and create streaming context
39
ray.init()
40
ctx = StreamingContext.Builder().build()
41
42
# Create a simple streaming pipeline
43
ctx.read_text_file("input.txt") \
44
.set_parallelism(1) \
45
.flat_map(lambda x: x.split()) \
46
.map(lambda x: (x, 1)) \
47
.key_by(lambda x: x[0]) \
48
.reduce(lambda old_value, new_value:
49
(old_value[0], old_value[1] + new_value[1])) \
50
.filter(lambda x: "ray" not in x) \
51
.sink(lambda x: print("result", x))
52
53
# Submit the job
54
ctx.submit("word_count_job")
55
```
56
57
### Creating Data Streams from Collections
58
59
```python
60
from ray.streaming import StreamingContext
61
62
ctx = StreamingContext.Builder().build()
63
64
# Create stream from values
65
ctx.from_values("a", "b", "c") \
66
.map(lambda x: x.upper()) \
67
.sink(lambda x: print(x))
68
69
# Create stream from collection
70
data = [1, 2, 3, 4, 5]
71
ctx.from_collection(data) \
72
.filter(lambda x: x % 2 == 0) \
73
.map(lambda x: x * 2) \
74
.sink(lambda x: print(f"Even doubled: {x}"))
75
76
ctx.submit("collection_processing")
77
```
78
79
## Architecture
80
81
Ray Streaming implements a master-worker architecture using Ray actors:
82
83
### Core Components
84
- **StreamingContext**: Main entry point for creating and configuring streaming jobs
85
- **DataStream**: Represents a stream of data elements that can be transformed
86
- **StreamSource**: Entry points for data ingestion into the streaming pipeline
87
- **Operators**: Transformation functions (map, filter, reduce, etc.) applied to streams
88
- **Internal Runtime**: Java-based execution engine handling distributed processing, fault tolerance, and resource management
89
90
### Fault Tolerance
91
Ray Streaming provides automatic fault tolerance through:
92
- Periodic checkpointing of operator state
93
- Single-node failover that only restarts failed components
94
- Automatic recovery from the last successful checkpoint
95
- Message replay from upstream operators during recovery
96
97
## Capabilities
98
99
### Streaming Context and Job Management
100
101
Main entry point for Ray streaming functionality with job lifecycle management.
102
103
```python { .api }
104
class StreamingContext:
105
class Builder:
106
def option(self, key=None, value=None, conf=None) -> Builder
107
def build(self) -> StreamingContext
108
109
def source(self, source_func: SourceFunction) -> StreamSource
110
def from_values(self, *values) -> StreamSource
111
def from_collection(self, values) -> StreamSource
112
def read_text_file(self, filename: str) -> StreamSource
113
def submit(self, job_name: str) -> None
114
```
115
116
[→ Streaming Context Documentation](./streaming-context.md)
117
118
### Data Streams and Transformations
119
120
Stream transformation operations and data flow management.
121
122
```python { .api }
123
class DataStream:
124
def map(self, func) -> DataStream
125
def flat_map(self, func) -> DataStream
126
def filter(self, func) -> DataStream
127
def key_by(self, func) -> KeyDataStream
128
def union(self, *others) -> DataStream
129
def sink(self, func) -> DataStreamSink
130
def set_parallelism(self, parallelism: int) -> DataStream
131
```
132
133
[→ Data Streams Documentation](./data-streams.md)
134
135
### Source Functions and Data Ingestion
136
137
Data source implementations and custom source function interfaces.
138
139
```python { .api }
140
class SourceFunction:
141
def init(self, parallel_id: int, num_parallel: int) -> None
142
def fetch(self, collector) -> None
143
144
class CollectionSourceFunction(SourceFunction):
145
def __init__(self, values)
146
147
class LocalFileSourceFunction(SourceFunction):
148
def __init__(self, filename: str)
149
```
150
151
[→ Source Functions Documentation](./source-functions.md)
152
153
### Stream Processing Operations
154
155
Core stream transformation operators and windowing functions.
156
157
```python { .api }
158
# Transformation Operations
159
DataStream.map(func) # One-to-one transformation
160
DataStream.flat_map(func) # One-to-many transformation
161
DataStream.filter(func) # Element filtering
162
DataStream.union(*others) # Stream union
163
164
# Keyed Operations
165
KeyDataStream.reduce(func) # Stateful reduction
166
KeyDataStream.window(window_func) # Windowing operations
167
```
168
169
[→ Stream Operations Documentation](./stream-operations.md)
170
171
### Cross-Language Integration
172
173
Support for mixed Python/Java streaming applications.
174
175
```python { .api }
176
# Convert to Java stream for Java operators
177
DataStream.as_java_stream()
178
179
# Use Java operators with class names
180
java_stream.map("com.example.MyMapper")
181
java_stream.filter("com.example.MyFilter")
182
183
# Convert back to Python stream
184
java_stream.as_python_stream()
185
```
186
187
[→ Cross-Language Support Documentation](./cross-language.md)
188
189
## Advanced Features
190
191
### Custom Source Functions
192
193
Create custom data sources by implementing the SourceFunction interface:
194
195
```python
196
from ray.streaming.function import SourceFunction
197
198
class CustomSourceFunction(SourceFunction):
199
def init(self, parallel_id, num_parallel):
200
self.counter = 0
201
self.max_count = 1000
202
203
def fetch(self, collector):
204
if self.counter < self.max_count:
205
collector.collect(f"message-{self.counter}")
206
self.counter += 1
207
else:
208
# Signal end of stream
209
collector.close()
210
```
211
212
### Configuration Options
213
214
Configure streaming jobs with various options:
215
216
```python
217
ctx = StreamingContext.Builder() \
218
.option("streaming.worker-num", "4") \
219
.option("streaming.context.backend.type", "MEMORY") \
220
.option("streaming.checkpoint.interval", "5000") \
221
.build()
222
```
223
224
### Error Handling and Monitoring
225
226
Ray Streaming integrates with Ray's monitoring and error handling:
227
- Built-in metrics collection and reporting
228
- Integration with Ray Dashboard for job monitoring
229
- Automatic error recovery and notification
230
- Configurable retry policies and failure handling
231
232
## Getting Started
233
234
1. **Install Ray**: `pip install ray`
235
2. **Initialize Ray**: `ray.init()`
236
3. **Create Context**: `ctx = StreamingContext.Builder().build()`
237
4. **Build Pipeline**: Chain stream operations using fluent API
238
5. **Submit Job**: `ctx.submit("job_name")`
239
240
For detailed examples and API references, see the individual capability documentation pages linked above.