Python中Kafka流连接的实现策略与实践

import os
from datetime import timedelta
from quixstreams import Application

# 假设Kafka broker地址和Quix平台配置已通过环境变量设置
# 例如：os.environ["Quix__Sdk__Token"] = "YOUR_QUIX_TOKEN"
#      os.environ["Quix__Broker__Address"] = "YOUR_BROKER_ADDRESS"
#      os.environ["Quix__Workspace__Id"] = "YOUR_WORKSPACE_ID"

# 1. 初始化Quix Application
# app = Application.Quix("my-join-app") # 生产环境建议使用更完整的配置

# 假设我们已经初始化了应用，并定义了输入和输出主题
# 为了演示，我们使用占位符
app = Application.Quix("manual-join-example")
input_topic_orders = app.topic("orders-topic", value_deserializer="json") # 订单流
input_topic_customers = app.topic("customers-topic", value_deserializer="json") # 客户信息流
output_topic_joined = app.topic("joined-orders-customers", value_serializer="json") # 连接后的输出流

# 获取一个用于存储连接状态的共享状态存储
# 状态存储是持久化的，可以在不同的窗口和处理实例间共享
join_state_store = app.get_state_store("join-data-store")

def update_and_check_join(key: str, message_value: dict, stream_type: str) -> dict or None:
    """
    更新共享状态存储并尝试执行连接。
    此函数将在处理每个消息时被调用。
    """
    # 存储当前消息到状态存储中，以键为前缀，区分来源
    # 例如：'order-key123' -> {'order_id': '123', 'product': 'A'}
    #       'customer-key123' -> {'customer_id': '123', 'name': 'John Doe'}
    join_state_store.set(f"{stream_type}-{key}", message_value)

    # 尝试从状态存储中获取另一个流的匹配数据
    partner_stream_type = "customer" if stream_type == "order" else "order"
    partner_data = join_state_store.get(f"{partner_stream_type}-{key}")

    joined_result = None
    if partner_data:
        # 如果找到匹配项，执行连接逻辑
        if stream_type == "order":
            joined_result = {
                "order_data": message_value,
                "customer_data": partner_data,
                "join_key": key
            }
        else: # stream_type == "customer"
            joined_result = {
                "order_data": partner_data,
                "customer_data": message_value,
                "join_key": key
            }

        # 成功连接后，可以选择从状态存储中清除这些键，避免重复连接
        # 这对于一次性连接非常有用，但如果需要多次连接或更新，则需要更复杂的逻辑
        join_state_store.delete(f"order-{key}")
        join_state_store.delete(f"customer-{key}")

    return joined_result

def process_streams(stream_manager):
    # 处理订单流
    stream_manager.topic(input_topic_orders).hopping_window(
        time_span=timedelta(seconds=10), # 窗口持续时间
        interval=timedelta(seconds=5),   # 窗口跳动间隔
    ).reduce(
        # reduce函数将消息累积到窗口的局部状态中，并在此处触发连接检查
        # 对于每个消息，我们调用 update_and_check_join
        lambda current_window_state, message: (
            # 这里的 current_window_state 可以用来累积窗口内的连接结果
            # 但为了简化，我们直接在每次消息处理时尝试连接并返回结果
            current_window_state.update({"latest_join_result": update_and_check_join(message.key, message.value, "order")}) or current_window_state
        ),
        initial_value={}
    ).to_topic(output_topic_joined, 
               lambda _, window_state: window_state.get("latest_join_result")
                                       if window_state.get("latest_join_result") else None)

    # 处理客户信息流
    stream_manager.topic(input_topic_customers).hopping_window(
        time_span=timedelta(seconds=10),
        interval=timedelta(seconds=5),
    ).reduce(
        lambda current_window_state, message: (
            current_window_state.update({"latest_join_result": update_and_check_join(message.key, message.value, "customer")}) or current_window_state
        ),
        initial_value={}
    ).to_topic(output_topic_joined, 
               lambda _, window_state: window_state.get("latest_join_result")
                                       if window_state.get("latest_join_result") else None)

# 运行应用程序
# if __name__ == "__main__":
#     print("Starting Quix Streams application for manual join...")
#     app.run(process_streams)
#     print("Quix Streams application stopped.")