Ray 與 Kafka 整合開發高效能分散式資料流處理系統

import json
from confluent_kafka import Producer
import ray

@ray.remote
class KafkaProducer:
    def __init__(self, broker: str = 'localhost:9092'):
        conf = {'bootstrap.servers': broker}
        self.producer = Producer(**conf)

    def produce(self, data: dict, key: str = None, topic: str = 'test'):
        def delivery_callback(err, msg):
            if err:
                print('Message failed delivery: ', err)
            else:
                print(f"Message delivered to topic {msg.topic()} " +
                      f"partition {msg.partition()} offset {msg.offset()}")

        binary_key = key.encode('UTF8') if key is not None else None
        self.producer.produce(topic=topic, value=json.dumps(data).encode('UTF8'),
                              key=binary_key, callback=delivery_callback)
        self.producer.poll(0)

    def destroy(self):
        self.producer.flush(30)

#### 內容解密：
在這段程式碼中，我們首先定義了一個 `KafkaProducer` 類別，並使用 `@ray.remote` 裝飾器將其轉換為 Ray 的遠端 Actor。在 `__init__` 方法中，我們初始化了一個 Kafka Producer，並組態了 Bootstrap Server。

`produce` 方法負責將資料寫入到 Kafka Topic 中。它接受一個字典型態的資料、一個可選的 Key 和一個 Topic 名稱。我們選擇使用字典來表示資料，因為它是一種通用且容易轉換為 JSON 的格式。

`delivery_callback` 方法用於檢查訊息是否成功寫入到 Kafka 中。如果訊息寫入失敗，我們會列印錯誤訊息；如果成功，則列印成功訊息。

最後，`destroy` 方法在應用程式離開時被呼叫，它等待最多 30 秒以確保所有未送出的訊息都已經被送出。

from confluent_kafka import Consumer
from uuid import uuid4
import ray

@ray.remote
class KafkaConsumer:
    def __init__(self, callback, group: str = 'ray', broker: str = 'localhost:9092',
                 topic: str = 'test', restart: str = 'latest'):
        consumer_conf = {
            'bootstrap.servers': broker,
            'group.id': group,
            'session.timeout.ms': 6000,
            'auto.offset.reset': restart
        }
        self.consumer = Consumer(consumer_conf)
        self.topic = topic
        self.id = str(uuid4())
        self.callback = callback

    def start(self):
        self.run = True
        def print_assignment(consumer, partitions):
            print(f'Consumer {self.id}')
            print(f'Assignment: {partitions}')
        self.consumer.subscribe([self.topic], on_assign=print_assignment)
        while self.run:
            msg = self.consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                print(f"Consumer error: {msg.error()}")
            else:
                self.callback(self.id, msg)

    def stop(self):
        self.run = False

    def destroy(self):
        self.consumer.close()

import ray

def process_message(id, message):
    print(f"Received message from {id}: {message.value().decode('utf-8')}")

# 初始化 Ray
ray.init()

# 建立 Kafka Producer 和 Consumer Actor
producer = KafkaProducer.remote()
consumer = KafkaConsumer.remote(process_message)

# 啟動 Consumer Actor
consumer.start.remote()

# 傳送一些測試資料到 Kafka Topic
for i in range(10):
    producer.produce.remote({'data': f'message {i}'})

# 停止 Consumer Actor
consumer.stop.remote()

# 銷毀 Producer 和 Consumer Actor
producer.destroy.remote()
consumer.destroy.remote()

# 關閉 Ray
ray.shutdown()

def setup_topics(broker: str = 'localhost:9092', topics: list = ['test'],
                 partitions: int = 10, replication: int = 1):
    # 刪除現有主題
    def wait_for_operation_completion(futures: dict, success: str, failure: str):
        for topic, f in futures.items():
            try:
                f.result()  # The result itself is None
                print(f"Topic {topic} {success}")
            except Exception as e:
                print(f"{failure} {topic} error {e}")

    admin = AdminClient({'bootstrap.servers': broker})
    fs = admin.delete_topics(topics)
    wait_for_operation_completion(fs, " is deleted", "Failed to delete topic ")
    sleep(3)

    # 建立新主題
    new_topics = [NewTopic(topic, num_partitions=partitions,
                           replication_factor=replication) for topic in topics]
    fs = admin.create_topics(new_topics)
    wait_for_operation_completion(fs, " is created", "Failed to create topic ")

# 單純的回撥函式來印出訊息
def print_message(consumer_id: str, msg):
    print(f"Consumer {consumer_id} 新訊息: topic={msg.topic()} "
          f"partition= {msg.partition()} offset={msg.offset()} "
          f"key={msg.key().decode('UTF8')}")
    print(json.loads(msg.value().decode('UTF8')))

# 初始化資料並啟動 Ray
setup_topics()
seed(1)
ray.init()

# 啟動消費者和生產者
n_consumers = 1  # 消費者數量
consumers = [KafkaConsumer.remote(print_message) for _ in range(n_consumers)]
producer = KafkaProducer.remote()
refs = [c.start.remote() for c in consumers]

# 釋出訊息
user_name = 'john'
user_favorite_color = 'blue'

# 永遠迴圈釋出訊息到 Kafka
try:
    while True:
        user = {
            'name': user_name,
            'favorite_color': user_favorite_color,
            'favorite_number': randint(0, 1000)
        }
        producer.produce.remote(user, str(randint(0, 100)))
        sleep(1)

# 優雅終止
except KeyboardInterrupt:
    for c in consumers:
        c.stop.remote()
finally:
    for c in consumers:
        c.destroy.remote()
    producer.destroy.remote()
    ray.kill(producer)