data = range(200)
ray.init()

rx.from_(data).pipe(
    rxray.distribute(
        lambda: rx.pipe(ops.map(lambda i: i*2)),
    ),
).subscribe()

data = [(i, j) for i in range(17) for j in range(100)]
random.shuffle(data)
ray.init()

rx.from_(data).pipe(
    rxray.distribute(
        lambda: rx.pipe(
            ops.group_by(lambda i: i[0]),
            ops.flat_map(lambda g: g.pipe(
                ops.map(lambda i: i[1]),
                ops.average(),
                ops.map(lambda i: (g.key, i)),
            ))
        ),
        actor_selector=rxray.partition_by_key(lambda i: i[0]),
    ),
).subscribe()