Pathway 实时数据处理框架：流批一体的现代数据处理引擎

# 基本安装
pip install pathway

# 安装带有所有连接器的完整版本
pip install pathway[complete]

# 安装特定连接器
pip install pathway[kafka,postgres,s3]

# 拉取Pathway Docker镜像
docker pull pathwaycom/pathway:latest

# 运行Pathway容器
docker run -it --rm pathwaycom/pathway:latest

# 克隆仓库
git clone https://github.com/pathwaycom/pathway.git
cd pathway

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate  # Windows

# 安装依赖
pip install -e "[dev,test,doc]"

# pathway_config.yaml

# 执行器配置
executor:
  num_threads: 4
  max_memory: 8GB

# 状态管理配置
state:
  storage: "local"
  path: "./state"
  checkpoint_interval: 60s

# 连接器配置
connectors:
  kafka:
    bootstrap_servers: "localhost:9092"
    consumer_group: "pathway-consumer"
  postgres:
    host: "localhost"
    port: 5432
    user: "pathway"
    password: "password"
    database: "pathway_db"

# basic_example.py
import pathway as pw

# 1. 定义数据源（批处理示例）
# 从CSV文件读取数据
table = pw.io.csv.read(
    "data/input.csv",
    schema={
        "id": int,
        "name": str,
        "value": float,
        "timestamp": pw.Type.datetime(
            format="%Y-%m-%d %H:%M:%S"
        )
    }
)

# 2. 定义数据处理逻辑
transformed = table.filter(lambda row: row.value > 100)
aggregated = transformed.groupby(pw.this.name).reduce(
    name=pw.this.name,
    count=pw.reducers.count(),
    sum_value=pw.reducers.sum(pw.this.value)
)

# 3. 定义数据输出（批处理示例）
pw.io.csv.write(aggregated, "data/output.csv")

# 4. 运行处理流程
if __name__ == "__main__":
    pw.run()

# stream_example.py
import pathway as pw
from datetime import timedelta

# 1. 定义Kafka数据源
kafka_input = pw.io.kafka.read(
    brokers=["localhost:9092"],
    topics=["input_topic"],
    group_id="pathway-group",
    schema={
        "id": int,
        "value": float,
        "timestamp": pw.Type.datetime(
            format="%Y-%m-%d %H:%M:%S"
        )
    },
    # 配置实时处理
    autocommit_duration_ms=1000
)

# 2. 定义实时处理逻辑
# 窗口聚合：每5分钟计算一次平均值
windowed = kafka_input.windowby(
    pw.this.timestamp, 
    window=timedelta(minutes=5),
    delay=timedelta(minutes=1)
).reduce(
    window_start=pw.reducers.min(pw.this.timestamp),
    window_end=pw.reducers.max(pw.this.timestamp),
    avg_value=pw.reducers.avg(pw.this.value),
    count=pw.reducers.count()
)

# 3. 定义数据输出到PostgreSQL
pw.io.postgres.write(
    windowed,
    table_name="windowed_results",
    dsn="postgresql://pathway:password@localhost:5432/pathway_db"
)

# 4. 运行实时处理流程
if __name__ == "__main__":
    pw.run()

# batch_stream_unified.py
import pathway as pw
from datetime import timedelta

# 1. 定义统一数据源接口
def get_data_source(is_streaming: bool):
    if is_streaming:
        # 流数据源：Kafka
        return pw.io.kafka.read(
            brokers=["localhost:9092"],
            topics=["input_topic"],
            group_id="pathway-group",
            schema={"id": int, "value": float, "category": str}
        )
    else:
        # 批数据源：CSV文件
        return pw.io.csv.read(
            "data/historical_data.csv",
            schema={"id": int, "value": float, "category": str}
        )

# 2. 定义统一处理逻辑
def process_data(data_table):
    # 过滤
    filtered = data_table.filter(lambda row: row.value > 50)
    
    # 分组聚合
    aggregated = filtered.groupby(pw.this.category).reduce(
        category=pw.this.category,
        count=pw.reducers.count(),
        sum_value=pw.reducers.sum(pw.this.value),
        avg_value=pw.reducers.avg(pw.this.value)
    )
    
    return aggregated

# 3. 定义统一输出接口
def write_results(results_table, is_streaming: bool):
    if is_streaming:
        # 流输出：PostgreSQL
        pw.io.postgres.write(
            results_table,
            table_name="real_time_results",
            dsn="postgresql://pathway:password@localhost:5432/pathway_db"
        )
    else:
        # 批输出：CSV文件
        pw.io.csv.write(results_table, "data/batch_results.csv")

# 4. 运行处理流程
if __name__ == "__main__":
    # 选择处理模式
    STREAMING_MODE = True  # 设置为False切换到批处理模式
    
    # 统一处理流程
    data = get_data_source(STREAMING_MODE)
    results = process_data(data)
    write_results(results, STREAMING_MODE)
    
    pw.run()

# real_time_ml.py
import pathway as pw
from sklearn.linear_model import LinearRegression
import numpy as np

# 1. 定义数据源
training_data = pw.io.csv.read(
    "data/training_data.csv",
    schema={"x1": float, "x2": float, "y": float}
)

streaming_data = pw.io.kafka.read(
    brokers=["localhost:9092"],
    topics=["inference_topic"],
    group_id="ml-group",
    schema={"x1": float, "x2": float, "id": int}
)

# 2. 训练机器学习模型
# 将Pathway表转换为NumPy数组
train_X = np.array(list(training_data.select(pw.this.x1, pw.this.x2)))
train_y = np.array(list(training_data.select(pw.this.y)))

# 训练线性回归模型
model = LinearRegression()
model.fit(train_X, train_y)

# 3. 定义实时推理函数
def predict(row):
    X = np.array([[row.x1, row.x2]])
    return model.predict(X)[0]

# 4. 应用模型进行实时推理
predictions = streaming_data.select(
    pw.this.id,
    pw.this.x1,
    pw.this.x2,
    prediction=pw.apply(predict, pw.this)
)

# 5. 输出结果
pw.io.postgres.write(
    predictions,
    table_name="predictions",
    dsn="postgresql://pathway:password@localhost:5432/pathway_db"
)

# 6. 运行实时推理流程
if __name__ == "__main__":
    pw.run()

# 示例：配置并行处理
import pathway as pw

# 设置执行器线程数
pw.Config.set_execution_parallelism(8)

# 其他优化配置
pw.Config.set_max_memory_usage("16GB")
pw.Config.set_checkpoint_interval("30s")

# performance_test.py
import pathway as pw
import time
import random
import csv
from datetime import datetime

# 1. 生成测试数据
def generate_test_data(num_records):
    """生成测试数据"""
    data = []
    for i in range(num_records):
        data.append({
            "id": i,
            "value": random.uniform(0, 1000),
            "category": random.choice(["A", "B", "C", "D", "E"]),
            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
        })
    return data

# 2. 批量处理性能测试
def test_batch_performance():
    """测试批处理性能"""
    # 生成不同大小的测试数据
    test_sizes = [1000, 10000, 100000, 1000000]
    results = []
    
    for size in test_sizes:
        print(f"测试批处理大小: {size}")
        
        # 生成测试数据文件
        test_data = generate_test_data(size)
        with open("test_batch_data.csv", "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=test_data[0].keys())
            writer.writeheader()
            writer.writerows(test_data)
        
        # 执行批处理
        start_time = time.time()
        
        # 读取数据
        table = pw.io.csv.read(
            "test_batch_data.csv",
            schema={
                "id": int,
                "value": float,
                "category": str,
                "timestamp": pw.Type. datetime(format="%Y-%m-%d %H:%M:%S.%f")
            }
        )
        
        # 处理数据
        processed = table.filter(lambda row: row.value > 500)
        aggregated = processed.groupby(pw.this.category).reduce(
            category=pw.this.category,
            count=pw.reducers.count(),
            sum_value=pw.reducers.sum(pw.this.value),
            avg_value=pw.reducers.avg(pw.this.value)
        )
        
        # 写入结果
        pw.io.csv.write(aggregated, "test_batch_result.csv")
        
        # 运行处理
        pw.run()
        
        end_time = time.time()
        duration = end_time - start_time
        throughput = size / duration
        
        print(f"  处理时间: {duration:.2f}秒")
        print(f"  吞吐量: {throughput:.2f}条/秒")
        
        results.append({
            "size": size,
            "duration": duration,
            "throughput": throughput,
            "mode": "batch"
        })
    
    return results

# 3. 流处理性能测试
def test_stream_performance():
    """测试流处理性能"""
    # 测试不同的流速率
    stream_rates = [100, 500, 1000, 5000, 10000]
    test_duration = 30  # 测试持续时间（秒）
    results = []
    
    for rate in stream_rates:
        print(f"测试流速率: {rate}条/秒")
        
        # 模拟Kafka流数据
        class MockKafkaSource(pw.io.Source):
            def __init__(self, rate, duration):
                self.rate = rate
                self.duration = duration
                self.start_time = time.time()
                
            def read(self):
                elapsed = time.time() - self.start_time
                if elapsed > self.duration:
                    return []
                
                # 生成当前批次的数据
                batch_size = int(self.rate * 0.1)  # 每100ms生成一批
                data = []
                for i in range(batch_size):
                    data.append({
                        "id": int(elapsed * self.rate + i),
                        "value": random.uniform(0, 1000),
                        "category": random.choice(["A", "B", "C", "D", "E"])
                    })
                
                time.sleep(0.1)  # 控制批次间隔
                return data
        
        # 执行流处理
        start_time = time.time()
        
        # 创建模拟数据源
        source = MockKafkaSource(rate, test_duration)
        table = pw.io.custom.read(source, schema={"id": int, "value": float, "category": str})
        
        # 处理数据
        processed = table.filter(lambda row: row.value > 500)
        aggregated = processed.groupby(pw.this.category).reduce(
            category=pw.this.category,
            count=pw.reducers.count(),
            sum_value=pw.reducers.sum(pw.this.value)
        )
        
        # 写入结果（这里使用内存存储）
        results_store = []
        def write_result(row):
            results_store.append(row)
        
        pw.io.custom.write(aggregated, write_result)
        
        # 运行处理
        pw.run()
        
        end_time = time.time()
        actual_duration = end_time - start_time
        processed_records = len(results_store) * 5  # 估计处理的总记录数
        throughput = processed_records / actual_duration
        
        print(f"  实际处理时间: {actual_duration:.2f}秒")
        print(f"  处理记录数: {processed_records}")
        print(f"  吞吐量: {throughput:.2f}条/秒")
        
        results.append({
            "rate": rate,
            "duration": actual_duration,
            "processed_records": processed_records,
            "throughput": throughput,
            "mode": "stream"
        })
    
    return results

# 4. 运行性能测试
if __name__ == "__main__":
    print("=== 批处理性能测试 ===")
    batch_results = test_batch_performance()
    
    print("\n=== 流处理性能测试 ===")
    stream_results = test_stream_performance()
    
    # 保存测试结果
    with open("performance_results.csv", "w", newline="") as f:
        fieldnames = ["mode", "size/rate", "duration", "throughput", "processed_records"]
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        
        for result in batch_results:
            writer.writerow({
                "mode": result["mode"],
                "size/rate": result["size"],
                "duration": result["duration"],
                "throughput": result["throughput"],
                "processed_records": result["size"]
            })
        
        for result in stream_results:
            writer.writerow({
                "mode": result["mode"],
                "size/rate": result["rate"],
                "duration": result["duration"],
                "throughput": result["throughput"],
                "processed_records": result["processed_records"]
            })
    
    print("\n性能测试完成，结果已保存到 performance_results.csv")