ストリームとは何か？

Copy
データ = 集合（構造）

Copy
# データ構造
numbers = [1, 2, 3, 4, 5]

result = []

# 処理
# 新しいデータ構造
for n in numbers:
    if n % 2 == 0:
        result.append(n * 2)

print(result)

Copy
データ構造
↓
処理
↓
新しいデータ構造

Copy
データを保存してから処理する

Copy
データ = 流れ

Copy
numbers = [1, 2, 3, 4, 5]

# 処理の仕組みを作る
stream = (n * 2 for n in numbers if n % 2 == 0)

# データを流す
for v in stream:
    print(v)

Copy
1 → 条件チェック → skip
2 → 条件チェック → ×2 → 4
3 → 条件チェック → skip
4 → 条件チェック → ×2 → 8
5 → 条件チェック → skip

Copy
処理の仕組みを先に作る
↓
データが流れてくる
↓
その処理が実行される

Copy
データを保持してから処理するのではなく
流れてくるデータを逐次処理する

Copy
Source → Operator → Sink

Copy
# Source
numbers = [1,2,3,4,5]

# Operator
# filter: 偶数だけ残す
# map: 値を2倍する
stream = (n*2 for n in numbers if n%2==0)

# Sink
for v in stream:
    print(v)

Copy
lines = open("log.txt").readlines()

errors = []
for line in lines:
    if "ERROR" in line:
        errors.append(line)

Copy
ファイル
↓
すべてメモリに読み込む
↓
処理する

Copy
巨大ログ
大量データ
継続的なデータ

Copy
with open("log.txt") as f:
    for line in f:
        if "ERROR" in line:
            print(line)

Copy
データを1件読む
↓
処理する
↓
次のデータ

Copy
numbers = [1,2,3,4,5]

filtered = []
for n in numbers:
    if n % 2 == 0:
        filtered.append(n)

mapped = []
for n in filtered:
    mapped.append(n * 2)

print(mapped)

Copy
データ
↓
filter
↓
新しい配列
↓
map
↓
新しい配列

Copy
中間データ構造を何度も作る

Copy
numbers = [1,2,3,4,5]

# 処理パイプライン作成
stream = (n * 2 for n in numbers if n % 2 == 0)

# 処理パイプラインにデータを流す
for v in stream:
    print(v)

Copy
Source
↓
filter
↓
map
↓
Sink

Copy
処理の流れをパイプラインとして表現できる

Copy
ログ
ユーザーイベント
ネットワーク通信
センサーデータ

Copy
イベント
↓
イベント
↓
イベント

Copy
イベント発生
↓
処理
↓
次のイベント

Copy
Reactive Programming
Observable
Event Stream

Copy
Stream
 ├ データ取得方式
 │   ├ Pull
 │   └ Push
 │
 ├ 実行モデル
 │   ├ Sync
 │   └ Async
 │
 └ データの性質
     ├ Finite
     ├ Infinite
     └ Event

Copy
Python Generator
→ Pull + Sync + Finite

Angular Observable
→ Push + Async + Event

Copy
Pull
Push

Copy
Consumer → Source

Copy
Iterator
Generator
Java Stream

Copy
Source → Consumer

Copy
Observable
Event Stream

Copy
Sync
Async

Copy
処理
↓
次の処理
↓
次の処理

Copy
Python Generator
Java Stream API

Copy
イベント
↓
非同期処理
↓
次のイベント

Copy
Observable
Async Iterator
Async Generator

Copy
Finite
Infinite
Event

Copy
配列
ファイル
DB結果

Copy
ログ
センサー
リアルタイムデータ

Copy
クリックイベント
WebSocket
メッセージキュー

Copy
Sync                     Async
            -----------------------------------------
Pull        Python Generator         Async Iterator
            Java Stream

Push        実用的でない              RxJS Observable
            （Consumer処理中に        Event Stream
             Sourceがブロックされる）

Copy
Pull + Sync
Pull + Async
Push + Async

Copy
Python Generator
Java Stream API

Copy
numbers = [1,2,3,4,5]

stream = (n*2 for n in numbers if n%2==0)

for v in stream:
    print(v)

Copy
List<Integer> numbers = List.of(1,2,3,4,5);

numbers.stream()
       .filter(n -> n % 2 == 0)
       .map(n -> n * 2)
       .forEach(System.out::println);

Copy
Async Generator
Async Iterator

Copy
import asyncio

async def stream():
    for i in range(3):
        await asyncio.sleep(1)
        yield i

async def main():
    async for v in stream():
        print(v)

asyncio.run(main())

Copy
RxJS Observable
Event Stream

Copy
import { fromEvent } from 'rxjs';

const clicks = fromEvent(document, 'click');

clicks.subscribe(event => {
  console.log("click", event);
});

Copy
データを
集合ではなく
流れとして扱う
抽象モデル

Copy
Source → Operator → Sink

Copy
Pull / Push
Sync / Async
Finite / Infinite / Event

Copy
大量データ処理
ログ処理
イベント処理
リアクティブプログラミング