流处理、实时分析与 RAG 驱动的 Python ETL 框架：构建智能数据管道 (上)

工作流程分为索引（离线/近线）和检索生成（实时）。Python 在其中的地位无可替代，LangChain 和 LlamaIndex 是构建应用的事实标准，配合 Sentence-BERT 等嵌入模型和各类向量数据库客户端，能轻松搭建 RAG 链路。

RAG 与流处理的融合点在于：流处理管道实时更新向量库，流中的事件触发 RAG 查询，生成的文本再流式传输回管道。

智能 ETL 框架架构设计

设计目标与原则

我们的目标是构建一个实时、可扩展、弹性且智能化的框架。遵循分层解耦、事件驱动、状态管理显式化、API 优先和配置驱动的原则。关键是要确保组件松耦合，便于替换和升级。

分层架构蓝图

+-----------------------------------------------------------------------+
| 用户接口层 (UI/API)                                                   |
| - 实时仪表盘 (Grafana, Superset), 查询接口 (REST, GraphQL), 告警通知 |
+-----------------------------------------------------------------------+
^ (查询/订阅)
| v
+-----------------------------------------------------------------------+
| 服务与编排层 (Service & Orchestration)                                |
| - API 网关，RAG 服务，实时查询服务，工作流编排                        |
+-----------------------------------------------------------------------+
^ (请求/结果)
| v
+-----------------------------------------------------------------------+
| 实时分析层 (Real-Time Analytics)                                      |
| - 流处理引擎，实时分析库，CEP 引擎，状态后端                          |
+-----------------------------------------------------------------------+
^ (处理结果/状态查询)
| v
+-----------------------------------------------------------------------+
| 存储层 (Storage)                                                      |
| - 消息队列，向量数据库，实时数据库，对象存储，关系型/NoSQL DB         |
+-----------------------------------------------------------------------+
^ (原始数据/知识源)
| v
+-----------------------------------------------------------------------+
| 数据源层 (Data Sources)                                               |
| - 流数据源，CDC，消息队列，API/Webhooks，文件系统，知识库             |
+-----------------------------------------------------------------------+

核心模块详解

数据接入与缓冲层

这一层负责可靠捕获数据。连接器可以是原生 SDK、CDC 工具（如 Debezium）或通用协议接收器。消息队列方面，Kafka 仍是事实标准，Pulsar 和 RabbitMQ 也是不错的选择。数据格式推荐 JSON、Avro 或 Protobuf。

Python 实现时，利用 asyncio 处理高并发连接，集成 Schema Registry 管理 Schema，并监控接入延迟和积压量。

流处理引擎层

这是框架的计算核心。根据第二章的分析选择合适的引擎。状态管理是关键，需配置状态后端（如 RocksDB）并定义 TTL。

下面是一个使用 PySpark Structured Streaming 处理用户活动的示例。注意这里使用了 Watermark 来处理乱序事件，并通过窗口函数进行聚合。

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, window, countDistinct
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

spark = SparkSession.builder.appName("RealtimeUserActivity").getOrCreate()

# 定义 Schema (假设数据是 JSON 格式)
schema = StructType([
    StructField("user_id", StringType(), True),
    StructField("event_type", StringType(), True),
    StructField("page_url", StringType(), True),
    StructField("timestamp", TimestampType(), True)
])

# 从 Kafka 读取流
kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers","broker1:9092,broker2:9092") \
    .option("subscribe","user_activity") \
    .option("startingOffsets","latest") \
    .load()

# 解析 JSON 值
parsed_df = kafka_df.selectExpr("CAST(value AS STRING)") \
    .select(from_json("value", schema).alias("data")) \
    .select("data.*")

# 计算每分钟不同用户访问次数
minute_activity_df = parsed_df \
    .withWatermark("timestamp","5 minutes") \
    .groupBy(window(col("timestamp"),"1 minute"), col("event_type")) \
    .agg(countDistinct("user_id").alias("unique_users"))

# 检测特定事件模式 (简化版 CEP)
# 假设要检测用户连续三次登录失败
from pyspark.sql.functions import lag, count
from pyspark.sql.window import Window

login_failures_df = parsed_df.filter(col("event_type")=="login_failed")
window_spec = Window.partitionBy("user_id").orderBy("timestamp")
flagged_df = login_failures_df \
    .withColumn("prev_event_type", lag("event_type",1).over(window_spec)) \
    .withColumn("prev_prev_event_type", lag("event_type",2).over(window_spec)) \
    .filter((col("prev_event_type")=="login_failed")&(col("prev_prev_event_type")=="login_failed")) \
    .select("user_id","timestamp").distinct()

# 写入结果到控制台
query1 = minute_activity_df.writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()
query2 = flagged_df.writeStream \
    .outputMode("update") \
    .format("console") \
    .start()

spark.streams.awaitAnyTermination()

实时分析层

基于流处理结果执行更复杂的分析，如在线 ML。可以使用 river 库进行增量学习。流处理作业将特征数据发送给在线模型服务，模型返回预测结果。

from river import compose, linear_model, metrics, optim, preprocessing
from river import stream

# 模拟一个实时数据流
def data_stream():
    # ... 生成或获取实时特征 (X) 和标签 (y)
    yield ({'feature1': 0.5, 'feature2': 1.2}, True)

# 定义在线模型 (线性回归 + 标准化)
model = compose.Pipeline(
    preprocessing.StandardScaler(),
    linear_model.LinearRegression(optimizer=optim.SGD(0.01)))

# 评估指标
metric = metrics.MAE()

# 在线训练与预测
for x, y in data_stream():
    y_pred = model.predict_one(x)
    metric.update(y, y_pred)
    model.learn_one(x, y)
    print(f"MAE: {metric.get():.4f}")

向量存储与 RAG 引擎层

管理知识库的嵌入向量并提供相似性搜索。LlamaIndex 和 LangChain 是构建 RAG 应用的标准工具。

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.storage.storage_context import StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb

# 1. 离线/近线索引
# 加载文档
documents = SimpleDirectoryReader('./docs').load_data()

# 创建 Chroma 客户端和集合
chroma_client = chromadb.Client()
chroma_collection = chroma_client.get_or_create_collection("knowledge_base")

# 设置向量存储
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

# 创建索引
index = VectorStoreIndex.from_documents(
    documents, storage_context=storage_context,
    transformations=[SentenceSplitter(chunk_size=500, chunk_overlap=50)])

# 2. 实时 RAG 查询服务 (FastAPI 集成)
from fastapi import FastAPI
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.retrievers import VectorIndexRetriever

app = FastAPI()

@app.post("/rag_query")
async def rag_query(query: str):
    retriever = VectorIndexRetriever(index=index, similarity_top_k=3)
    query_engine = RetrieverQueryEngine.from_args(retriever)
    response = query_engine.query(query)
    return {"query": query, "response": str(response)}

服务与输出层

将结果通过标准接口暴露。FastAPI 是高性能 API 服务的首选，支持 REST、GraphQL 和 WebSocket。仪表盘可使用 Grafana 或 Superset。告警系统可集成 Prometheus Alertmanager 或 Slack Webhook。

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
import asyncio
import json

app = FastAPI()

# 模拟流式输出 (例如 RAG 生成过程)
async def generate_stream_response(query: str):
    words = ["This", " is", " a", " streamed", " response"]
    for word in words:
        yield f"data: {json.dumps({'token': word})}\n\n"
        await asyncio.sleep(0.2)
    yield "data: [DONE]\n\n"

@app.get("/stream_query")
async def stream_query(query: str):
    return StreamingResponse(generate_stream_response(query), media_type="text/event-stream")

监控与可观测性层

全面监控框架的运行状态。Prometheus 收集指标，ELK Stack 收集日志，OpenTelemetry 进行分布式追踪。

from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.exporter.jaeger.thrift import JaegerExporter

# 初始化 Tracing
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

jaeger_exporter = JaegerExporter(agent_host_name="jaeger", agent_port=6831)
span_processor = BatchSpanProcessor(jaeger_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)

@tracer.start_as_current_span("process_event")
def process_event(event):
    try:
        # ... 实际处理逻辑 ...
        return "processed"
    except Exception as e:
        span = trace.get_current_span()
        span.record_exception(e)
        raise