Data processing support in Ray

dask_on_ray.py

import ray
from ray.util.dask import ray_dask_get

import dask
import dask.dataframe as dd 
import pandas as pd
import numpy as np

dask.config.set(scheduler=ray_dask_get) # Sets Ray as the default backend.

ray.init()
df = pd.DataFrame(np.random.randint(0, 100, size=(2**10, 2**8)))
df = dd.from_pandas(df, npartitions=10)
print(df.head(10))

mars_on_ray.py

from mars.session import new_session
ray_session = new_session(backend=’ray’).as_default() # Set Ray as the default backend.

import mars.dataframe as md
import mars.tensor as mt

t = mt.random.randint(100, size=(2**10, 2**8))
df = md.DataFrame(t)
print(df.head(10).execute())

modin_on_ray.py

import ray
# Modin defaults to backing Ray’s object store with disk.
# Start Ray before importing modin to use shared memory instead.
ray.init()

import modin.pandas as pd
import numpy as np

frame_data = np.random.randint(0, 100, size=(2**10, 2**8))
df = pd.DataFrame(frame_data)
print(df.head(10))

pyspark_on_ray.py

import ray
import raydp

ray.init()

@ray.remote
class PySparkDriver:
    def __init__(self):
        self.spark = raydp.init_spark(
            app_name=”RayDP example”,
            num_executors=2,
            executor_cores=2,
            executor_memory=”4GB”)

    def foo(self):
        return self.spark.range(1000).repartition(10).count()

driver = PySparkDriver.remote()
print(ray.get(driver.foo.remote()))