Pathway 实时服务器日志监控实战：Filebeat + Kafka + Pathway 滑动窗口异常检测与告警

2026-09-07 15:40:18作者：丁柯新Fawn

本指南以 docs/2.developers/7.templates/ETL/7.realtime-log-monitoring.md 为骨架，结合仓库内完整示例工程讲解。日志是任何系统与应用的核心组件，承载着性能、使用情况与错误信息，但手工监控大规模日志既耗时又乏味。本文将带你用 Pathway 的数据处理框架对流式日志做事件驱动的实时分析——只要日志流入，结果自动更新。你将掌握两套可落地的架构：其一是在 ELK 栈的 Logstash 与 ElasticSearch 之间插入 Pathway 做滑动窗口统计并输出结果；其二是让 Pathway（经由 Kafka）直连 Filebeat，在检测到短窗口内连接数超阈值时把告警实时推送到 Slack 频道。阅读后可独立完成从 docker-compose 编排、日志生成到滑动窗口告警的完整链路搭建。

一个最小的实时告警方案（Short Version）

先看一个简单场景：当最近 1 秒内收到的日志超过 5 条时触发告警。借助 Pathway，你只需要在带时间戳列 timestamp 的表上计算一个滑动窗口，再把告警输出到指定 Slack 频道即可：

import pathway as pw
import requests
from datetime import timedelta

alert_threshold = 5
sliding_window_duration = timedelta(seconds=1)


SLACK_ALERT_CHANNEL_ID = "XXX"
SLACK_ALERT_TOKEN = "XXX"

rdkafka_settings = {
    "bootstrap.servers": "kafka:9092",
    "security.protocol": "plaintext",
    "group.id": "0",
    "session.timeout.ms": "6000",
}

inputSchema = pw.schema_builder(
    columns={
        "@timestamp": pw.column_definition(dtype=str),
        "message": pw.column_definition(dtype=str),
    }
)


# We use the Kafka connector to listen to the "logs" topic
# We only need the timestamp and the message
log_table = pw.io.kafka.read(
    rdkafka_settings,
    topic="logs",
    format="json",
    schema=inputSchema,
    autocommit_duration_ms=100,
)
log_table = log_table.select(timestamp=pw.this["@timestamp"], log=pw.this.message)
log_table = log_table.select(
    pw.this.log,
    timestamp=pw.this.timestamp.dt.strptime("%Y-%m-%dT%H:%M:%S.%fZ"),
)

# A sliding window is computed based on log_table using the timestamp
t_sliding_window = log_table.windowby(
    log_table.timestamp,
    window=pw.temporal.sliding(
        hop=timedelta(milliseconds=10), duration=sliding_window_duration
    ),
    behavior=pw.temporal.common_behavior(
        cutoff=timedelta(seconds=0.1),
        keep_results=False,
    ),
).reduce(timestamp=pw.this._pw_window_end, count=pw.reducers.count())

# There is an alert if there are more than alert_threshold logs in the window
t_alert = t_sliding_window.reduce(count=pw.reducers.max(pw.this.count)).select(
    alert=pw.this.count >= alert_threshold
)


# on_alert_event defines the behavior of the Slack channel when the alert is received
def on_alert_event(key, row, time, is_addition):
    alert_message = "Alert '{}' changed state to {}".format(
        row["alert"],
        "ACTIVE" if is_addition else "INACTIVE",
    )
    requests.post(
        "https://slack.com/api/chat.postMessage",
        data="text={}&channel={}".format(alert_message, SLACK_ALERT_CHANNEL_ID),
        headers={
            "Authorization": "Bearer {}".format(SLACK_ALERT_TOKEN),
            "Content-Type": "application/x-www-form-urlencoded",
        },
    ).raise_for_status()


# The alert table is connected to the Slack channel with the output connector
pw.io.subscribe(t_alert, on_alert_event)

time.sleep(5)
# Launching the computation.
pw.run()

整个链路只有四步：从 Kafka 读入日志 → 字符串时间转成 DateTime → windowby 开滑动窗口聚合计数 → 结果超阈值时把状态变化推给 Slack。下面展开每一步在生产架构里如何落地。

如何监控日志：两种目标架构

假设你有一台 Nginx 服务器，需要监控其访问日志，并已经部署了标准的 ELK 栈：用 Filebeat 采集日志，经 Logstash 转发到 ElasticSearch。现在你希望获得"更主动"的监控——每当服务器在 1 秒窗口内连接数超过 5 就触发一次告警。围绕在链路中插入 Pathway 的位置，本文定义了两个场景：

场景 #1：把 Pathway 插在 Logstash 与 ElasticSearch 之间。日志流经 Logstash → Kafka → Pathway（做实时统计/告警判定）→ ElasticSearch（存储统计结果）。
场景 #2：追求更极致的低延迟，去掉中间环节，让 Filebeat 直连（经 Kafka）Pathway，Pathway 直接向专属 Slack 频道推送告警。

两套架构在本仓库中均有可运行源码，路径为 examples/projects/realtime-log-monitoring/logstash-pathway-elastic（场景 #1）与 examples/projects/realtime-log-monitoring/filebeat-pathway-slack（场景 #2），两者的目录结构完全对应下文出现的配置与脚本。

需要说明的前提是：Pathway 目前没有 Filebeat/Logstash 专用连接器，消息统一通过一个 Kafka 实例中转。Kafka 输入连接器的完整说明见 Kafka 连接器指南。

用 Docker 容器编排全部服务

与其逐台机器安装 Filebeat、Logstash 等服务，不如全部容器化。Docker 把应用与依赖打包成可移植容器，保证在任何操作系统与配置下环境一致；docker-compose 则允许同时管理多个容器。在工程根目录放一个 docker-compose.yml，结构如下：

version: "3.7"
services:
  filebeat:
    build:
    environment:
    volumes:
  logstash:
    build:
    ...

每个应用（即一个 "service"）都在这里定义相关参数：build 下参数决定如何构建容器（基于哪个镜像等）。注意 filebeat 是服务名而非镜像名，实际镜像由 build 决定。

将 Pathway 接入你的服务器

第一步：配置 Filebeat 与 Logstash（场景 #1 的采集端）

为了让 Filebeat、Logstash 采集日志并经 Kafka 转发给 Pathway，需要同时在 docker-compose.yml 中配置这两个服务。仓库中完整的 compose 文件见 logstash-pathway-elastic/docker-compose.yml：

services:
  filebeat:
    build:
      context: .
      dockerfile: ./filebeat-src/Dockerfile
    links:
      - logstash:logstash
    depends_on:
      - logstash
  logstash:
    image: docker.elastic.co/logstash/logstash:8.6.2
    volumes:
    - ./logstash-src/logstash.conf:/usr/share/logstash/pipeline/logstash.conf
    ports:
      - 5044:5044

Filebeat 端需要一份 Dockerfile 加一份配置文件：Dockerfile 把配置文件拷入容器以设定 Filebeat 的输入/输出，另外还拷贝脚本 generate_input_stream.sh 用于生成人工日志流。仓库实现见 filebeat-src/Dockerfile：

FROM docker.elastic.co/beats/filebeat:8.6.1

COPY ./filebeat-src/filebeat.docker.yml /usr/share/filebeat/filebeat.yml
COPY ./filebeat-src/generate_input_stream.sh /usr/share/filebeat/generate_input_stream.sh
USER root

RUN mkdir /input_stream/
RUN touch /input_stream/example.log

RUN chown root:filebeat /usr/share/filebeat/filebeat.yml
RUN chmod go-w /usr/share/filebeat/filebeat.yml

配置文件用于指定要监控的目录（本例为 /input_stream/*）以及日志输出到哪里（这里是 Logstash）。仓库实现见 filebeat-src/filebeat.docker.yml：

filebeat.inputs:
- type: filestream
  id: my-logs
  paths:
    - /input_stream/*

filebeat.config.modules:
  path: /usr/share/filebeat/modules.d/
  reload.enable: false

output.logstash:
  enabled: true
  hosts: ["logstash:5044"]

为简化演示，示例没有加任何认证机制（如需开启，可参考 Filebeat 官方 Logstash 输出插件文档）。

Logstash 端负责把消息发往 Kafka，因此要使用 Logstash 的 Kafka 输出插件，并指定 topic 与消息格式。本例 topic 为 logs；由于 Filebeat 以 JSON 输出更新，Logstash 侧保持同样格式。仓库实现见 logstash-src/logstash.conf：

input {
  beats {
    port => 5044
  }
}

output {
    kafka {
        codec => json
        topic_id => "logs"
        bootstrap_servers => "kafka:9092"
        key_serializer => "org.apache.kafka.common.serialization.StringSerializer"
        value_serializer => "org.apache.kafka.common.serialization.StringSerializer"
    }
}

关键点在于：Logstash 的输入是 Filebeat（Beats 系列 agent 之一）并监听 5044 端口——这个端口在 Filebeat 配置的 output 段指定过，也必须同时在 docker-compose 中开放。输出目标是 Kafka，Kafka 在此扮演 Logstash 与 Pathway 之间的消息中转站。

生成人工日志

如果你没有真实服务器可监控，或流量不足以触发告警，可以用脚本在 Filebeat 监控的目录里生成人工数据。下面的脚本为模拟一次明显的流量尖峰：前 100 秒每秒产生 1 条日志，随后在 1 秒内产生 100 条。仓库实现见 generate_input_stream.sh：

#!/bin/bash

src="../../../input_stream/example.log"

sleep 1

for LOOP_ID in {1..100}
do
    printf "$LOOP_ID\n" >> $src
    sleep 1
done
for LOOP_ID in {101..200}
do
    printf "$LOOP_ID\n" >> $src
    sleep 0.01
done

该脚本应放在 ./filebeat-src/ 目录中，由 Dockerfile 拷入 Filebeat 容器。

关于 Nginx 日志

Nginx 是众多组织托管 Web 应用时常用的 Web 服务器与反向代理，监控 Nginx 日志是 Filebeat 的典型用例。上面的配置可直接用于 Nginx 日志——只需把 paths 指向 Nginx 日志所在目录即可；此外 Filebeat 还提供了 Nginx 专用 module 可做更精细的解析。

第二步：连接 Pathway

配置 Kafka 与 ZooKeeper

Pathway 与 Filebeat/Logstash 之间需要一个 Kafka 实例做网关；Kafka 依赖 ZooKeeper，因此两者都需要容器：

  zookeeper:
    image: confluentinc/cp-zookeeper:5.5.3
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
  kafka:
    image: confluentinc/cp-enterprise-kafka:5.5.3
    depends_on: [zookeeper]
    environment:
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_BROKER_ID: 1
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_JMX_PORT: 9991
    ports:
    - 9092:9092
    command: sh -c "((sleep 15 && kafka-topics --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 1 --topic logs)&) && /etc/confluent/docker/run "

command 一行会在各容器就绪后先创建 topic logs 再启动 Kafka；若要换 topic，改这行里的 logs 即可。

Pathway 管道容器

Pathway 没有独立官方镜像，这里用 Dockerfile 自建容器。compose 中指定所用 Dockerfile 并声明依赖顺序：

  pathway:
    build:
      context: .
      dockerfile: ./pathway-src/Dockerfile
    depends_on: [kafka, logstash]

Dockerfile 只需一个 Python 镜像并 pip install 即可，仓库实现见 pathway-src/Dockerfile：

FROM --platform=linux/x86_64 python:3.10

RUN pip install -U pathway
RUN pip install python-dateutil
COPY ./pathway-src/alerts.py alerts.py

CMD ["python", "-u", "alerts.py"]

⚠️ 出于兼容性考虑，这里固定使用 x86_64 架构的 Linux 容器。除了 pathway，还需要额外安装 python-dateutil 包。

Kafka 输入连接器。在 rdkafka_settings 之外，用 topic 参数指定监听 logs topic。随后需要声明输入数据 schema：Filebeat 生成、经 Logstash 转发的 JSON 中有用的两列是 @timestamp 与 message，其余是 Filebeat 附加的元数据（本例丢弃）。因为列名含特殊字符 @timestamp，应使用内联式 schema 定义（pw.schema_builder）而不是类定义；schema 定义方式的完整讲解见 Schema 指南。仓库实现见 pathway-src/alerts.py：

import pathway as pw

rdkafka_settings = {
    "bootstrap.servers": "kafka-server:9092",
    "security.protocol": "sasl_ssl",
    "sasl.mechanism": "SCRAM-SHA-256",
    "group.id": "$GROUP_NAME",
    "session.timeout.ms": "6000",
    "sasl.username": "username",
    "sasl.password": "password",
}

inputSchema = pw.schema_builder(
  columns={
    '@timestamp': pw.column_definition(dtype=str),
    'message': pw.column_definition(dtype=str)
  }
)

log_table = pw.io.kafka.read(
    rdkafka_settings,
    topic="logs",
    format="json",
    schema=inputSchema,
    autocommit_duration_ms=100,
)

注意：上文示例展示的是开启 SASL_SSL 认证的写法；本仓库配套 docker-compose 中的 Kafka 使用 plaintext 明文协议，因此 alerts.py 里的 bootstrap.servers 为 kafka:9092、security.protocol 为 plaintext。以本地 compose 实际配置为准。

得到表之后还要做两步预处理：把 @timestamp 重命名为 timestamp，以便使用 Pathway 的点号（dot）记法；Filebeat 的时间戳是 ISO 8601 字符串，需要转成 DateTime 才能方便地计算两条日志的时间差。Pathway 提供了 DateTime API 处理日期时间。两步转换只需几行：

log_table = log_table.select(timestamp=pw.this["@timestamp"], log=pw.this.message)
log_table = log_table.select(
    pw.this.log,
    timestamp=pw.this.timestamp.dt.strptime("%Y-%m-%dT%H:%M:%S.%fZ"),
)

log_table 至此已准备好被处理。

第三步：Filebeat 直连（场景 #2）

如果不需要 Logstash、想从 Filebeat（经 Kafka）直连 Pathway，只需改动 docker-compose 与 Filebeat 配置：Filebeat 服务不再 links/depends_on Logstash，输出从 Logstash 换成 Kafka。

# docker-compose.yml 中 filebeat 服务简化为
services:
  filebeat:
    build:
      context: .
      dockerfile: ./filebeat-src/Dockerfile

# filebeat-src/filebeat.docker.yml
filebeat.inputs:
- type: filestream
  id: my-logs
  paths:
    - /input_stream/*

filebeat.config.modules:
  path: /usr/share/filebeat/modules.d/
  reload.enable: false

output.kafka:
  enabled: true
  hosts: ["kafka:9092"]
  topic: "logs"
  group_id: 1
  ssl.enabled: false

同样不启用认证；如需认证可参考 Filebeat 官方 Kafka 输出插件文档。两个场景完整的 compose 可在仓库对照：filebeat-pathway-slack/docker-compose.yml（场景 #2，服务依赖收敛为 filebeat / kafka / zookeeper / pathway）。

用滑动窗口实时处理日志

窗口：流处理实时统计的基石

实时服务器监控这类流式实时统计，只关心最近的数据，如最近 10 分钟内的数据会被保留并用来检测随时间变化的异常与模式，这部分数据称为"窗口"。窗口本质是一个固定大小的缓冲，以固定间隔（通常滑动或重叠）在数据流上移动，持续采集并处理固定数量的数据或固定时长的数据。主流窗口技术分两类：

滚动窗口（Tumbling windows）：把数据流切分为固定大小、互不重叠的时间区间；窗口装满即关闭并开启新窗口。
滑动窗口（Sliding windows）：以固定大小、互相重叠的方式在数据流上移动；每段数据在窗口走完后，窗口按固定步长向前推进。

实时监控通常优先选择滑动窗口：滚动窗口会把数据切成互不重叠的片段，一旦切点不对，可能恰好错过你想捕获的模式。滑动窗口在计算时刻纳入最近数据，其有效性高度依赖计算时机，而窗口起点通常由用户周期性设定。

Pathway 中的滑动窗口

Pathway 用 windowby 轻松构建滑动窗口，它会自动维护数据所需的窗口集合，并在新数据到达时增量更新。windowby 的另一大优势是内置**数据遗忘（data forgetting）**机制：通过指定 behavior，可以指示 Pathway 遗忘旧窗口及其数据，对日志监控场景有两个直接收益：

告警是否成立只基于近期窗口计算；
程序可以运行在恒定内存中，而不是保留全部历史数据。

关于 temporal behaviors 的系统性讲解见 Temporal Behaviors 用户指南。

windowby 需要 window 参数来定义窗口类型。日志监控用滑动窗口：每 10 毫秒（hop）生成一个、每个持续 1 秒（duration）。由于告警阈值只取决于日志条数，聚合用 count reducer：

t_sliding_window = log_table.windowby(
    log_table.timestamp,
    window=pw.temporal.sliding(
        hop=timedelta(milliseconds=10), duration=sliding_window_duration
    ),
    behavior=pw.temporal.common_behavior(
        cutoff=timedelta(seconds=0.1),
        keep_results=False,
    ),
).reduce(count=pw.reducers.count())

从源码看，pw.temporal.sliding 的实现定义在 python/pathway/stdlib/temporal/_window.py：参数 hop 为窗口产生频率、duration 为窗口长度，二者必须给出其一；也可以只给 ratio，用 duration = hop * ratio 替代，或用 origin 指定首窗起点。pw.temporal.common_behavior 定义在 python/pathway/stdlib/temporal/temporal_behavior.py，其参数含义为：

delay：相对于窗口起点延迟初始输出（None 为不启用延迟机制）；
cutoff：允许算子丢弃早于当前算子时间减去 cutoff 的输入/输出，从而释放内存；
keep_results：为 False 时，时间窗结果会在窗口过期后从输出表中移除，避免表无限膨胀。

最后构建一张只含单个布尔条目的告警表：取任意窗口中的日志最大条数，若 ≥ 阈值则置 alert 为 True。

t_alert = t_sliding_window.reduce(count=pw.reducers.max(pw.this.count))
t_alert = t_alert.select(alert=pw.this.count >= alert_threshold)

每当新数据到达，各表都会被更新：早于 cutoff 的旧窗口被剔除，保留条数超过阈值时告警置为真。这一点在仓库的镜像测试 python/pathway/tests/examples/realtime-log-slack.py 中有直观体现——它把 10 条带时间戳的日志喂入同样的 schema 与时间解析逻辑，验证 5 条阈值下告警状态的翻转。该测试文件头部注释还特别说明：它必须与 examples/projects/realtime-log-monitoring/filebeat-pathway-slack/pathway-src/alerts.py 保持同步，这从侧面印证了文档代码即仓库中的可运行示例。

输出结果

借助输出连接器，你可以把结果发往任意目标存储：把流量统计发给 ElasticSearch，或把告警消息直接推给 Slack 换取更快的响应。

场景 #1：把数据写入 ElasticSearch

先起一个 ElasticSearch 容器（仓库见 logstash-pathway-elastic/docker-compose.yml）：

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.6.2
    ports:
      - 9200:9200
    environment:
      - discovery.type=single-node
      - "ES_JAVA_OPTS=-Xms2g -Xmx2g"
      - ELASTIC_PASSWORD=password
      - xpack.security.enabled=false
    restart: unless-stopped

默认用户名为 elastic。无需手工建索引——Pathway 向 ElasticSearch 写数据时会自动创建。容器就绪几秒后即可查询：

curl localhost:9200/alerts/_search?pretty

该命令展示 alerts 索引收到的消息，查看其他索引时替换名字即可。

Pathway 侧用 ElasticSearch 输出连接器写表：

pw.io.elasticsearch.write(
    t_alert,
    "http://elasticsearch:9200",
    auth=pw.io.elasticsearch.ElasticSearchAuth.basic("elastic", "password"),
    index_name="alerts",
)

注意此处不能用 localhost，而要用 docker-compose 的服务名 elasticsearch。

场景 #2：把告警发到 Slack

出于响应速度考虑，你可能希望告警直达 Slack 频道。需要频道 ID 与 token：

import requests

def on_alert_event(key, row, time, is_addition):
    alert_message = "Alert '{}' changed state to {}".format(
        row["alert"],
        "ACTIVE" if is_addition else "INACTIVE",
    )
    requests.post(
        "https://slack.com/api/chat.postMessage",
        data="text={}&channel={}".format(alert_message, SLACK_ALERT_CHANNEL_ID),
        headers={
            "Authorization": "Bearer {}".format(SLACK_ALERT_TOKEN),
            "Content-Type": "application/x-www-form-urlencoded",
        },
    ).raise_for_status()


pw.io.subscribe(t_alert, on_alert_event)

这里用 pw.io.subscribe 把告警表 t_alert 的每次变化（is_addition=True 表示新增/激活，False 表示撤销/失效）回调给 on_alert_event。回调需要 requests 包，须在 Dockerfile 中安装，见 pathway-src/Dockerfile：

FROM --platform=linux/x86_64 python:3.10

RUN pip install -U pathway
RUN pip install requests
RUN pip install python-dateutil
COPY ./pathway-src/alerts.py alerts.py

CMD ["python", "-u", "alerts.py"]

这样你的告警就会被直接送到 Slack 频道。

把所有组件拼起来

场景 #1 的完整布局

场景 #1（Logstash + ElasticSearch）的目录结构如下（仓库对应 logstash-pathway-elastic）：

.
├── filebeat-src/
│   ├── Dockerfile
│   ├── filebeat.docker.yml
│   └── generate_input_stream.sh
├── logstash-src/
│   └── logstash.conf
├── pathway-src/
│   ├── alerts.py
│   └── Dockerfile
├── docker-compose.yml
└── Makefile

完整的 Pathway 实现（与仓库 pathway-src/alerts.py 一致）：

import time
from datetime import timedelta
import pathway as pw

alert_threshold = 5
sliding_window_duration = timedelta(seconds=1)

rdkafka_settings = {
    "bootstrap.servers": "kafka:9092",
    "security.protocol": "plaintext",
    "group.id": "0",
    "session.timeout.ms": "6000",
}

inputSchema = pw.schema_builder(
  columns={
    '@timestamp': pw.column_definition(dtype=str),
    'message': pw.column_definition(dtype=str)
  }
)

log_table = pw.io.kafka.read(
    rdkafka_settings,
    topic="logs",
    format="json",
    schema=inputSchema,
    autocommit_duration_ms=100,
)
log_table = log_table.select(timestamp=pw.this["@timestamp"], log=pw.this.message)
log_table = log_table.select(
    pw.this.log,
    timestamp=pw.this.timestamp.dt.strptime("%Y-%m-%dT%H:%M:%S.%fZ").dt.timestamp(),
)

t_sliding_window = log_table.windowby(
    log_table.timestamp,
    window=pw.temporal.sliding(
        hop=timedelta(milliseconds=10), duration=sliding_window_duration
    ),
    behavior=pw.temporal.common_behavior(
        cutoff=timedelta(seconds=0.1),
        keep_results=False,
    ),
).reduce(timestamp=pw.this._pw_window_end, count=pw.reducers.count())

t_alert = t_sliding_window.reduce(count=pw.reducers.max(pw.this.count)).select(
    alert=pw.this.count >= alert_threshold
)

pw.io.elasticsearch.write(
    t_alert,
    "http://elasticsearch:9200",
    auth=pw.io.elasticsearch.ElasticSearchAuth.basic("elastic", "password"),
    index_name="alerts_logs",
)

time.sleep(5)

pw.run()

你可能注意到 pw.run() 前有一个 time.sleep(5)：这是为了等待 Kafka 就绪。不加也能运行，但会在日志里看到连接 Kafka 失败的报错。同样千万别忘了 pw.run()——没有它，任何计算都不会执行。

场景 #2 的完整布局

场景 #2（仅 Filebeat，告警直达 Slack）的目录结构（仓库对应 filebeat-pathway-slack）：

.
├── filebeat-src/
│   ├── Dockerfile
│   ├── filebeat.docker.yml
│   └── generate_input_stream.sh
├── pathway-src/
│   ├── alerts.py
│   └── Dockerfile
├── docker-compose.yml
└── Makefile

从连接 Kafka 到把告警转发给 Slack 的完整实现（与仓库 pathway-src/alerts.py 一致）：

import time
from datetime import timedelta
import pathway as pw
import requests

alert_threshold = 5
sliding_window_duration = timedelta(seconds=1)

SLACK_ALERT_CHANNEL_ID = "XXX"
SLACK_ALERT_TOKEN = "XXX"

rdkafka_settings = {
    "bootstrap.servers": "kafka:9092",
    "security.protocol": "plaintext",
    "group.id": "0",
    "session.timeout.ms": "6000",
}

inputSchema = pw.schema_builder(
  columns={
    '@timestamp': pw.column_definition(dtype=str),
    'message': pw.column_definition(dtype=str)
  }
)


log_table = pw.io.kafka.read(
    rdkafka_settings,
    topic="logs",
    format="json",
    schema=inputSchema,
    autocommit_duration_ms=100,
)
log_table = log_table.select(timestamp=pw.this["@timestamp"], log=pw.this.message)
log_table = log_table.select(
    pw.this.log,
    timestamp=pw.this.timestamp.dt.strptime("%Y-%m-%dT%H:%M:%S.%fZ"),
)

t_sliding_window = log_table.windowby(
    log_table.timestamp,
    window=pw.temporal.sliding(
        hop=timedelta(milliseconds=10), duration=sliding_window_duration
    ),
    behavior=pw.temporal.common_behavior(
        cutoff=timedelta(seconds=0.1),
        keep_results=False,
    ),
).reduce(timestamp=pw.this._pw_window_end, count=pw.reducers.count())

t_alert = t_sliding_window.reduce(count=pw.reducers.max(pw.this.count)).select(
    alert=pw.this.count >= alert_threshold
)

def on_alert_event(key, row, time, is_addition):
    alert_message = "Alert '{}' changed state to {}".format(
        row["alert"],
        "ACTIVE" if is_addition else "INACTIVE",
    )
    requests.post(
        "https://slack.com/api/chat.postMessage",
        data="text={}&channel={}".format(alert_message, SLACK_ALERT_CHANNEL_ID),
        headers={
            "Authorization": "Bearer {}".format(SLACK_ALERT_TOKEN),
            "Content-Type": "application/x-www-form-urlencoded",
        },
    ).raise_for_status()

pw.io.subscribe(t_alert, on_alert_event)

time.sleep(5)

pw.run()

仓库中的两个示例还带有一行 pw.set_license_key(...)：运行 Pathway 企业级功能（Scale）时需要 license key；若只用 Community 版可注释掉该行。文档示例为了可读性省略了它。

Makefile：一键启停与调试

启动、停止、进入容器可用下面的 Makefile（仓库见各场景目录下的 Makefile）：

build:
	docker-compose up -d

stop:
	docker-compose down -v

connect:
	docker-compose exec filebeat bash
connect-pathway:
	docker-compose exec pathway bash

启动 generate_input_stream.sh 时，先用 make connect 进入 Filebeat 容器，再执行：

./generate_input_stream.sh

这会生成一条人工日志流。随后就能在 ElasticSearch 实例或 Slack 上（取决于所选场景）看到它的效果：脚本会在约 100 秒内保持低频日志，然后瞬间灌入上百条日志，1 秒内 > 5 条 的阈值被击穿，告警随之触发。

结语

至此，你已经能对服务器日志做实时监控。流式数据的实时统计分析是实时流处理的要害，而传统窗口技术存在一个本质局限：它们按固定周期触发，对事件本身无感知。在流式数据面前，这种设定要么浪费资源，要么损失精度。

Pathway 的做法是事件驱动的事件窗口（event-based windows）：窗口是"增量更新"而非"从头重算"的，每来一条新事件，框架自动更新受影响的窗口与下游结果——滑动窗口始终基于最新数据计算，既不会漏掉任何数据点，也不会把同一窗口重复算两遍。整条管道的运行方式可以概括为：定义好管道后，Pathway 替你打理所有更新；你要做的，只是安坐屏幕前，等告警自己跳到 Slack 里。若想进一步探索，可对照 first-realtime-app 指南了解实时应用的通用写法，或在 examples/notebooks/tutorials/windows_temporal_behavior.ipynb 中亲手实验窗口与时间行为的差异。

pathway

Python ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.

项目地址：https://gitcode.com/GitHub_Trending/pa/pathway

登录后查看全文