edoakes

import asyncio

from ray import serve

@serve.deployment
class MessageConsumer:
    def __init__(self, topic: str):
        asyncio.get_running_loop().create_task(
            self.poll_for_messages(topic)
        )

edoakes

import argparse
import time
import random
import math

parser = argparse.ArgumentParser(description="Approximate digits of Pi using Monte Carlo simulation.")
parser.add_argument("--num-samples", type=int, default=1000000)
parser.add_argument("--parallel", default=False, action="store_true")
parser.add_argument("--distributed", default=False, action="store_true")

edoakes

@pipeline.step
def preprocess(_input: str) -> PreprocessOutput:
    pass

@pipeline.step(num_replicas=10, num_gpus=1)
class Model1:
    def __call__(self, _input: PreprocessOutput) -> Model1Output:
	pass

@pipeline.step(num_replicas=5, num_cpus=1)

edoakes

import dash
from dash import dcc, html
from dash.dependencies import Input, Output
import pandas as pd
import plotly.graph_objs as obj
import uvicorn as uvicorn
from fastapi import FastAPI
from starlette.middleware.wsgi import WSGIMiddleware

import ray

edoakes

import dash
from dash import dcc, html
from dash.dependencies import Input, Output
import pandas as pd
import plotly.graph_objs as obj
import uvicorn as uvicorn
from fastapi import FastAPI
from starlette.middleware.wsgi import WSGIMiddleware

app = dash.Dash(__name__, requests_pathname_prefix="/dash/")

edoakes

from transformers import pipeline

@serve.deployment(route_prefix="/sentiment", name="sentiment")
class SentimentDeployment:
    def __init__(self):
        self.classifier = pipeline("sentiment-analysis")

    async def __call__(self, request):
        data = await request.body()
        [result] = self.classifier(str(data))

edoakes

import ray
from ray import serve

# Connect to the running Ray Serve instance.
ray.init(address='auto', namespace="serve-example", ignore_reinit_error=True)
serve.start(detached=True)

# Deploy the model.
SentimentDeployment.deploy()

edoakes

import joblib
import s3fs
import sklearn

@serve.deployment(route_prefix="/sentiment", name="sentiment-deployment")
class SentimentDeployment:
    def __init__(self):
        fs = s3fs.S3FileSystem(anon=True)
        with fs.open('ray-serve-blog/unigram_vectorizer.joblib', 'rb') as f:
            self.vectorizer = joblib.load(f)

edoakes

import ray
from ray import serve

ray.init(address='auto', namespace="serve-example", ignore_reinit_error=True)
serve.start(detached=True)

SentimentDeployment.deploy()

edoakes

import ray
from ray import serve
ray.init(address='auto', namespace="serve-example", ignore_reinit_error=True) # Connect to the local running Ray cluster.
serve.start(detached=True) # Start the Ray Serve processes within the Ray cluster.