背景

大批量 SERP 数据需要实时处理 (监控 + 告警 + 数据湖),用 Kafka 作为 SERP 数据流的核心基础设施最自然。

1. 准备

# 启动 Kafka(本地)
docker run -d --name kafka -p 9092:9092 \
  -e KAFKA_ADVERTISED_HOST=localhost \
  apache/kafka

# 安装 Python client
pip install kafka-python requests

2. Producer:抓 SERP → 推 Kafka

import json
import requests
from kafka import KafkaProducer
import time

producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
)

def fetch_and_produce():
    """调 serpbase 抓 SERP → 推 Kafka"""
    queries = ["python async", "rust tutorial", "react hooks"]

    for q in queries:
        r = requests.post(
            "https://api.serpbase.dev/google/search",
            headers={"X-API-Key": "sk_xxx"},
            json={"q": q, "gl": "us", "num": 10},
            timeout=10,
        )
        data = r.json()

        for i, item in enumerate(data.get("organic", []), 1):
            producer.send("serp-results", value={
                "query": q,
                "rank": i,
                "title": item["title"],
                "link": item["link"],
                "ts": int(time.time()),
            })
        producer.flush()

fetch_and_produce()

3. Consumer:实时处理 SERP 事件

import json
from kafka import KafkaConsumer
import requests

consumer = KafkaConsumer(
    "serp-results",
    bootstrap_servers="localhost:9092",
    value_deserializer=lambda v: json.loads(v.decode("utf-8")),
)

def process_event(event):
    """每个 SERP 事件做实时处理"""
    # 1. 实时分析(LLM 摘要)
    if is_important_keyword(event["query"]):
        summary = call_llm(event)
        send_slack(f"SERP 变化:{summary}")

    # 2. 写数据库
    db.serp_results.insert(event)

    # 3. 推下游(告警、数据湖)
    if event["rank"] > 10:
        alert_low_rank(event)

    if event["rank"] < 3:
        alert_top_rank(event)

    # 推送到 Elasticsearch
    es.index(index="serp-data", body=event)

for event in consumer:
    process_event(event)

4. 5 个工程细节

细节 1:Schema Registry

# 用 Avro / Protobuf 强 schema
{
  "type": "record",
  "name": "SerpResult",
  "fields": [
    {"name": "query", "type": "string"},
    {"name": "rank", "type": "int"},
    {"name": "title", "type": "string"},
    {"name": "ts", "type": "long"},
  ]
}

细节 2:exactly-once 语义

producer = KafkaProducer(
    transactional_id="serp-producer-1",  # 启用事务
    enable_idempotence=True,
)

细节 3:Consumer Group

consumer = KafkaConsumer(
    "serp-results",
    group_id="serp-processors",  # 多个 consumer 自动分片
)

细节 4:Avro 序列化 (节省 30% 空间)

import fastavro

# 写
with open("schema.avsc") as f:
    schema = json.load(f)
parsed = fastavro.parse_schema(schema)
buf = io.BytesIO()
fastavro.schemaless_writer(buf, parsed).write(record)
producer.send("serp-results", value=buf.getvalue())

细节 5:Kafka Streams 实时处理

// Java Kafka Streams 实时聚合
StreamsBuilder builder = new StreamsBuilder();
KStream<String, SerpResult> source = builder.stream("serp-results");

KTable<String, Long> rankByQuery = source
    .groupBy((k, v) -> v.query)
    .aggregate(
        () -> 0L,
        (k, v, acc) -> (long) v.rank,
        Materialized.as("rank-by-query")
    );

rankByQuery.toStream().to("rank-summary");

5. 5 个实战功能

功能 1:实时 SERP 排名监控

consumer = KafkaConsumer("serp-results", group_id="rank-monitor")
for event in consumer:
    if event.rank > 10:
        send_alert(event.query, event.rank)

功能 2:数据湖 (归档)

# 实时推 Kafka → 5 分钟归档到 S3
from kafka import KafkaConsumer

consumer = KafkaConsumer("serp-results", group_id="archiver")
batch = []
for event in consumer:
    batch.append(event)
    if len(batch) >= 1000:
        s3.put_object(Bucket="serp-data-lake", Key=f"raw/{int(time.time()) // 300 * 300}.json", Body=json.dumps(batch))
        batch = []

功能 3:Kafka Streams 实时聚合

KStream<String, SerpResult> events = builder.stream("serp-results");

KTable<Windowed<String>, Long> rankByHour = events
    .groupBy((k, v) -> v.query)
    .windowedBy(TimeWindows.of(Duration.ofHours(1)))
    .aggregate(
        () -> 0L,
        (k, v, acc) -> Math.min(acc, v.rank),
        Materialized.as("min-rank-by-hour")
    );

功能 4:实时 AI 处理

from kafka import KafkaConsumer
import asyncio

async def process_event(event):
    summary = await call_llm_async(event)
    await send_slack_async(summary)

consumer = KafkaConsumer("serp-results", group_id="ai-processors")
for event in consumer:
    asyncio.create_task(process_event(event))

功能 5:Kafka Connect → ClickHouse / Postgres

{
  "name": "serp-events-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "topics": "serp-results",
    "connection.url": "jdbc:postgresql://localhost:5432/seo",
    "auto.create": true,
  }
}

6. 实战数据 (1 个月)

7. 与 RabbitMQ / Redis Streams 对比

8. 5 个最佳实践

1. 用 Schema Registry(Avro / Protobuf)
2. 启用 idempotence(防止重复消息)
3. Consumer Group(自动分片)
4. Kafka Connect(标准化集成)
5. 监控(Kafka 自身 + 业务指标)

小结

serpbase + Kafka 流式管道:

• Producer 推 SERP 数据到 Kafka
• Consumer 实时处理 (告警 / 数据湖 / AI)
• Kafka Streams 聚合分析
• 月处理 100K 消息,$3 月成本

Kafka 适合大数据 + 高吞吐,serpbase + Kafka 解决 1k-10k QPS 的实时处理需求。