import ray

ray.init()  # 初始化 Ray 集群

@ray.remote
def add(x, y):
    return x + y

# 提交任務(wù)
result_id = add.remote(1, 2)
# 獲取任務(wù)結(jié)果
result = ray.get(result_id)
print(result)  # 輸出: 3

@ray.remote
class Counter:
    def __init__(self):
        self.value = 0

    def increment(self):
        self.value += 1
        return self.value

# 創(chuàng)建 Actor 實(shí)例
counter = Counter.remote()
# 調(diào)用 Actor 方法
for _ in range(5):
    print(ray.get(counter.increment.remote()))  # 輸出: 1, 2, 3, 4, 5

data = [1, 2, 3, 4, 5]
data_id = ray.put(data)  # 將數(shù)據(jù)存儲(chǔ)到對(duì)象存儲(chǔ)
retrieved_data = ray.get(data_id)  # 從對(duì)象存儲(chǔ)檢索數(shù)據(jù)
print(retrieved_data)  # 輸出: [1, 2, 3, 4, 5]

@ray.remote(num_gpus=1)  # 請(qǐng)求 1 個(gè) GPU
def train_model(data):
    # 模型訓(xùn)練代碼
    pass

import ray
import time

ray.init()

@ray.remote
def compute_heavy_task(i):
    time.sleep(1)  # 模擬耗時(shí)計(jì)算
    return i * i

# 提交多個(gè)任務(wù)
result_ids = [compute_heavy_task.remote(i) for i in range(10)]
# 獲取結(jié)果
results = ray.get(result_ids)
print(results)  # 輸出: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

import ray
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset

ray.init()

# 定義簡(jiǎn)單模型
class SimpleModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.linear = nn.Linear(10, 1)

    def forward(self, x):
        return self.linear(x)

# 模擬數(shù)據(jù)
X = torch.randn(1000, 10)
y = torch.randn(1000, 1)
dataset = TensorDataset(X, y)
dataloader = DataLoader(dataset, batch_size=32)

# 分布式訓(xùn)練函數(shù)
@ray.remote
def train_epoch(model_weight_id, dataloader_id, epoch):
    model_weight = ray.get(model_weight_id)
    model = SimpleModel()
    model.load_state_dict(model_weight)
    criterion = nn.MSELoss()
    optimizer = optim.SGD(model.parameters(), lr=0.01)
    dataloader = ray.get(dataloader_id)

    for batch_X, batch_y in dataloader:
        optimizer.zero_grad()
        outputs = model(batch_X)
        loss = criterion(outputs, batch_y)
        loss.backward()
        optimizer.step()

    # 返回更新后的模型權(quán)重
    updated_weight = model.state_dict()
    return updated_weight

# 初始化模型權(quán)重
model = SimpleModel()
initial_weight = model.state_dict()
weight_id = ray.put(initial_weight)
dataloader_id = ray.put(dataloader)

# 提交多個(gè)訓(xùn)練任務(wù)
for epoch in range(5):
    weight_id = train_epoch.remote(weight_id, dataloader_id, epoch)

# 獲取最終模型權(quán)重
final_weight = ray.get(weight_id)
print("Training completed!")

import ray
from confluent_kafka import Producer, Consumer

ray.init()

@ray.remote
class KafkaProducerActor:
    def __init__(self, bootstrap_servers):
        self.producer = Producer({'bootstrap.servers': bootstrap_servers})

    def produce(self, topic, message):
        self.producer.produce(topic, value=message)
        self.producer.flush()

@ray.remote
class KafkaConsumerActor:
    def __init__(self, bootstrap_servers, group_id, topic):
        self.consumer = Consumer({
            'bootstrap.servers': bootstrap_servers,
            'group.id': group_id,
            'auto.offset.reset': 'earliest'
        })
        self.consumer.subscribe([topic])

    def consume(self):
        msg = self.consumer.poll(1.0)
        if msg is not None and msg.error() is None:
            return msg.value().decode('utf-8')
        return None

# 創(chuàng)建生產(chǎn)者和消費(fèi)者 Actor
producer_actor = KafkaProducerActor.remote('localhost:9092')
consumer_actor = KafkaConsumerActor.remote('localhost:9092', 'test-group', 'test-topic')

# 生產(chǎn)消息
ray.get(producer_actor.produce.remote('test-topic', b'Hello, Ray!'))

# 消費(fèi)消息
message = ray.get(consumer_actor.consume.remote())
print(f"Received message: {message}")  # 輸出: Received message: Hello, Ray!

import ray

ray.init()  # 初始化本地 Ray 集群