Graylog本身自带Elasticsearch作为存储后端，但在一些轻量场景下，比如内网小规模日志采集、告警字段二次加工、或者需要把提取出来的结构化字段导出到本地做报表，直接把Graylog的提取结果写入SQLite会是一个性价比很高的方案。SQLite零部署、单文件、SQL支持完整，配合Graylog的Extractor机制，可以把散落在原始日志里的IP、用户名、错误码等字段抽出来，落成一张可以随意查询的表。这篇文章就围绕这个思路，完整走一遍从日志进入Graylog、字段提取、再到SQLite入库的全过程。
一、Graylog提取器的工作原理与配置
Graylog收到日志之后，原始内容存放在message字段里。如果想让系统识别出其中的关键信息，需要在Input上配置Extractor（提取器）。Extractor本质上是一段在消息进入管道时执行的解析逻辑，支持正则提取、JSON解析、Grok模式、分割符切割等多种方式。以一条典型的Nginx访问日志为例：
192.168.1.23 - - [12/Mar/2025:10:31:05 +0800] "GET /api/user HTTP/1.1" 200 512 "-" "Mozilla/5.0"
这条日志里包含客户端IP、请求方法、路径、状态码、响应大小，全挤在一个字符串里。我们可以在Graylog的Web界面中，进入对应Input的Manage extractors页面，新建一个Regular expression类型的Extractor，Source field选择message，用正则分组把各部分抽出来，分别命名为src_ip、http_method、request_path、status_code、response_size。配置完成后，每条进入该Input的消息都会自动携带这些新字段，搜索时可以直接写status_code:500这样的查询条件，不用再对message做全文匹配。
需要注意的一点是，正则提取有性能开销。如果日志量很大，建议优先使用JSON类型的Extractor，也就是在日志产生端（比如应用程序通过logback、winston输出）就把字段组织成JSON，Graylog只需要一次JSONPath解析即可，既稳定又高效。正则Extractor更适合处理那些无法改造输出格式的存量系统日志。
二、用Pipeline规则做更灵活的字段加工
Extractor适合做单条规则的字段抽取，如果需要条件判断、字段拼接或者数据清洗，Graylog的Pipeline规则更合适。Pipeline是一段用特定语法写的处理脚本，在消息流经管道时执行。例如我们想把状态码大于等于400的请求打上错误标签，并对src_ip做规范化处理，可以编写如下规则：
rule "mark error requests"
when
has_field("status_code") && to_long($message.status_code) >= 400
then
set_field("is_error", true);
set_field("error_level", to_long($message.status_code) >= 500 ? "server_error" : "client_error");
end这段规则做了两件事：一是给消息增加布尔字段is_error，二是根据状态码区间细分错误类型。规则写好后，需要把它关联到一个Stage，再把Pipeline挂到对应的Stream上。这样一来，只有匹配该Stream的消息才会执行规则，避免无关日志浪费计算资源。
Pipeline还支持Lookup Table，这是连接SQLite的关键桥梁。Lookup Table可以配置一个SQLite数据缓存适配器，让Graylog在处理消息时查外部数据表。典型用法是IP归属地查询：准备一张SQLite表存IP段与地区的映射，消息进来时用src_ip去查表，把结果写进新字段region。这样后续在Graylog里就能直接按地区维度做统计了。
三、把提取字段写入SQLite的完整实现
落库这一步有两种常见做法。第一种是利用Graylog的输出插件，通过GELF UDP输出把消息转发给一个自写的小服务，由该服务负责写SQLite。第二种更简单，直接写一个Python消费者，通过Graylog的REST API定时拉取数据。下面给出一个用Python将提取字段写入SQLite的示例：
import sqlite3
import json
import requests
# 创建本地数据库和表
conn = sqlite3.connect("graylog_extract.db")
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS access_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT,
src_ip TEXT,
http_method TEXT,
request_path TEXT,
status_code INTEGER,
response_size INTEGER,
is_error INTEGER,
region TEXT
)
""")
conn.commit()
# 从Graylog REST API拉取最近一小时日志
url = "http://192.168.0.1:9000/api/search/universal/relative"
params = {"query": "status_code:*", "range": 3600, "fields": "timestamp,src_ip,http_method,request_path,status_code,response_size,is_error,region"}
resp = requests.get(url, params=params, auth=("admin", "password"))
for msg in resp.json().get("messages", []):
f = msg.get("message", {})
cursor.execute(
"INSERT INTO access_log (ts, src_ip,http_method, request_path, status_code, response_size, is_error) VALUES (?,?,?,?,?,?,?)",
(f.get("timestamp"), f.get("src_ip"), f.get("http_method"),
f.get("request_path"), f.get("status_code"), f.get("response_size"),
1 if f.get("is_error") else 0)
)
conn.commit()
conn.close()数据一旦进了SQLite，后续的分析就完全交给SQL。比如统计最近一小时各错误类型的数据量、找出访问量最大的前十个IP，都是一条查询语句的事：
-- 按状态码分组统计错误请求数 SELECT status_code, COUNT(*) AS cnt FROM access_log WHERE is_error = 1 GROUP BY status_code ORDER BY cnt DESC; -- 访问量前十的客户端IP SELECT src_ip, COUNT(*) AS total, SUM(response_size) AS bytes FROM access_log GROUP BY src_ip ORDER BY total DESC LIMIT 10;
四、性能优化与常见坑
这套方案落地时有几个容易踩的坑。首先是写入频率，SQLite对并发写不友好，同一时刻只允许一个写事务，如果日志拉取间隔太短、数据量大，会出现database is locked报错。解决办法有两个：一是开启WAL模式，执行PRAGMA journal_mode=WAL;，允许读写一定程度并行；二是把INSERT改成批量提交，攒够一批再写，配合executemany效率提升明显。
其次是字段缺失问题。Graylog的Extractor如果匹配失败，消息里就不会有对应字段，Python端用f.get()取值时会拿到None，直接写入NOT NULL约束的列会报错。建议建表时放宽约束，或者在代码里对关键字段做默认值兜底，比如状态码缺失时统一记为0，便于后期排查。
最后是索引设计。access_log表数据量增长很快，查询常用的status_code、src_ip、ts三个字段建议建立索引，否则几百万行之后全表扫描会非常慢。同时可以定期归档，比如把七天以前的数据导出到单独的归档库文件，主库保持轻量。整体来看，Graylog负责采集与提取，SQLite负责落地与分析，两者分工明确，对于日均百万级以内的日志场景，这套组合完全够用，而且维护成本远低于重型数据库方案。