【免费下载链接】xerjXERJ is the new way for AI to search data. Its autoindex capability activates agents to know your data without the token waste of grep and sed. One command indexes code, docs, logs and PDFs for search, RAG, security audits and agent memory, using 40x fewer tokens than grep. Elasticsearch compatible, so existing clients just work.项目地址https://gitcode.com/GitHub_Trending/xe/xerj点击查看免费下载本篇指南讲解 XERJ 的_ml/datafeeds连续异常检测能力在已有的按需评分接口_score之上用一个 datafeed 把检测器挂到实时索引上按固定frequency自动重评分只把新出现的异常 bucket 追加到结果存储中供仪表盘或告警循环轮询。读完本文你可以从零搭建一个CPU 尖峰自动发现的实时告警回路创建检测器与 datafeed、理解_start后同步跑一次 后台定时重评分的机制、掌握结果过滤/上限、停止与删除的完整生命周期并能直接运行仓库自带的端到端脚本复现全部结果。场景先理解 datafeed 解决什么问题按需的_score端点是你主动发起一次评分、一次性拿回整条时间线。但仪表盘或告警循环不想每次都去问——它希望引擎持续盯着一个实时索引把新出现的异常主动递过来。这就是datafeed。XERJ 提供了 Elasticsearch 兼容的_ml/datafeeds接口面一个 datafeed 通过job_id引用一个已存在的异常检测器detector它从一个或多个indices读取数据可附带任意 ESquery过滤运行在固定的frequency上_start之后XERJ 先立即同步评分一次随后由后台任务按frequency周期性重评分只把新标记的 bucket 追加到该 job 的结果中结果通过GET /_ml/anomaly_detectors/{job}/results/records读回。评分数学与按需路径完全一致——同一套透明、确定性的滑动窗口均值/标准差 z-score实现上两者共用同一个内部函数score_detector_records见 engine/crates/xerj-api/src/es_compat.rs只是它自己跑起来了。诚实的能力边界原文档原样继承评分就是_score那套确定性滑动窗口 z-score——非常适合这个指标相对它自己近期水平跳了这类场景而不是季节性/多变量模型。结果保存在内存中每个 job 上限约 1 万条记录后台评分器是每节点一个 tokio 任务结果和运行中的 datafeed 在节点重启后不会保留重新加载的 datafeed 恢复为stopped状态再次_start就是一次全新运行。这是单节点能力没有跨节点 datafeed 调度器。它正适合单节点上的实时仪表盘或自包含告警回路而不是集群化持久 ML 管线的替代品。本指南的场景cpu_datafeed索引每分钟采一个 CPU 读数。前 16 分钟稳定在 20% 左右第 12 分钟出现一个失控尖峰96%。我们创建检测器、挂上 datafeed 并启动它然后在 datafeed 运行期间第 20 分钟落地一个第二个尖峰88%观察 datafeed 自己把它找出来。1. 建索引并创建检测器时间字段必须是date类型指标字段必须是数值类型。检测器与按需配方中的对象完全相同对一分钟一桶的cpu取均值mean(cpu)。curl -XPUT localhost:9200/cpu_datafeed -H Content-Type: application/json -d { mappings: { properties: { timestamp: { type: date }, cpu: { type: double } }} } curl -XPUT localhost:9200/_ml/anomaly_detectors/cpu-spike \ -H Content-Type: application/json -d { source_index: cpu_datafeed, time_field: timestamp, function: mean, field: cpu, bucket_span: 1m }检测器的完整配置项可以从 engine/crates/xerj-api/src/state.rs 中的MlDetector结构体确认字段说明默认值source_index要分析的源索引字符串或索引模式必填time_field用于分桶的时间字段timestampfunction指标函数count/mean/min/max/sumavg是mean的别名必填count时可省略fieldfield指标函数作用的字段count时无需—bucket_span桶跨度ES 时间间隔语法1h、5m、1d必填anomaly_threshold偏离基线多少个标准差算异常3.0description可选的人类可读描述—其中function会映射到 XERJ 引擎的聚合名count → value_count、mean → avg等见 engine/crates/xerj-api/src/es_compat.rs 的ml_metric_agg。2. 创建 datafeeddatafeed 必须提供job_id且该检测器必须已存在否则返回404错误体为 ES 形状的resource_not_found_exception。其余字段都是可选的indices默认取检测器的source_indexquery默认match_allfrequency按 ES 时间值解析2s、5m……解析不到时回落到 60 秒且钳制到 1s。curl -XPUT localhost:9200/_ml/datafeeds/cpu-spike-feed \ -H Content-Type: application/json -d { job_id: cpu-spike, frequency: 2s }响应ES 形状的 datafeed 配置{ datafeed_id: cpu-spike-feed, job_id: cpu-spike, state: stopped, indices: [cpu_datafeed], query: { match_all: {} }, frequency: 2s }从源码看indices接受字符串或字符串数组并且兼容indexes拼写创建时如果indices为空会自动填充检测器的source_index若同 id 的 datafeed 已存在其仍在运行的后台评分任务会先被abort()避免残留孤儿任务。对应实现在 engine/crates/xerj-api/src/es_compat.rs 的put_ml_datafeed。datafeed 配置本身会持久化到data_dir/_ml/datafeeds.jsonstate.rs 的MlDatafeed::save_all所以重启后 datafeed配置还在但运行态会被重置——见下节。3. 启动先同步跑一遍然后挂上定时器_start做三件事见 engine/crates/xerj-api/src/es_compat.rs 的start_ml_datafeed把状态置为started重置去重游标last_scored_ms——一次_start就是一次全新运行同步执行一次评分让结果立即可查同时清空该 job 上一次运行的旧结果避免跨运行累积tokio::spawn一个后台任务每frequency秒重评分一次。curl -XPOST localhost:9200/_ml/datafeeds/cpu-spike-feed/_start # - {started: true}读取第一次同步评分产出的结果curl localhost:9200/_ml/anomaly_detectors/cpu-spike/results/records{ count: 1, records: [ { job_id: cpu-spike, result_type: record, timestamp: 1783296720000, timestamp_iso: 2026-07-06T00:12:00.000Z, actual: 96.0, typical: 20.25, expected: 20.25, std_dev: 1.1636866703140785, z_score: 65.09484205018445, record_score: 100.0, anomaly_score: 100.0, bucket_span: 1m, is_anomaly: true } ] }每条记录复用_score端点逐 bucket 输出的字段另加 ES 风格的record_score0–100 严重度同时镜像为anomaly_score、result_type标签和typicalES 对基线expected的称呼。这套映射在ml_anomaly_record中逐字段完成es_compat.rs。对上述固定序列这些数值是确定性的——下面的演示脚本打印的就是这份记录。评分细节对理解z_score 65.09这类数字很有用前 4 个桶作为基线预热MIN_BASELINE 4只入基线不评分记录中带insufficient_baseline说明基线只吸收正常桶异常桶不会回灌进基线避免一个尖峰污染后续期望值z-score 归一化到 0–100刚好越过阈值约为 50 分偏离越大单调趋近 100ml_anomaly_scorees_compat.rs标准差塌缩到 0 时有保护逻辑z 值会被钳制在±1e6内输出避免除零。4. 连续部分让 datafeed 自己抓到第二个尖峰现在往正在运行的索引里推入一个新尖峰——不碰 datafeed 本身curl -XPOST localhost:9200/_bulk?refreshtrue \ -H Content-Type: application/x-ndjson --data-binary $ {index:{_index:cpu_datafeed}} {timestamp:2026-07-06T00:20:00Z,cpu:88} 在一个frequency心跳之内后台评分器重新分桶、看到新桶并把它追加进来。几秒后再轮询curl localhost:9200/_ml/anomaly_detectors/cpu-spike/results/records # - count: 2 (00:12 的尖峰和新的 00:20 尖峰)datafeed 会跟踪它已经评过分的最高桶时间戳last_scored_ms游标所以每一遍只追加新的异常——00:12 的记录永远不会重复。从源码看datafeed_score_pass对每个评过的桶无论是否异常都会推进这个高水位游标评分失败比如索引尚未创建是尽力而为的——不产生新记录游标也不动es_compat.rs。关于新鲜度的一则说明原文档原样继承。XERJ 的读路径会按(query, dataset_version)缓存聚合结果而 version 只在 segment flush 时才递增——所以一个相同的 query 打到只写了未 flush内存中文档的索引上可能拿到缓存的旧结果。因为 datafeed 每个心跳都重复发同一个 query其评分器刻意请求绕过缓存的读取内部把fresh true传给score_detector_records确保总能看到最新桶。按需的_score端点行为不变仍可享受缓存。另外两个值得知道的实现事实权限规则随任务走每索引可见性守卫是 tokio task-local而tokio::spawn不继承 task-local。源码注释专门记录了这个问题后台评分器会在启动请求的 task 上捕获当时的可见性规则再重新安装保证datafeed 评分器能看到的恰好等于启动它的人能看到的——受限 principal 的 datafeed 对未授权索引会安静地评出 0 条记录而不是越权读取spawn_datafeed_taskes_compat.rs对应的回归测试见 engine/crates/xerj-api/tests/ml_datafeed_cannot_read_a_brain.rs。重启语义load_all加载 datafeed 注册表时会把state强制重置为stopped、游标清空state.rs因为后台任务和内存结果存储都不跨重启存活。配置持久在运行态归零重新_start即全新开始。5. 过滤与截断结果results/records支持record_score下限query 参数或 JSON body 均可和size上限# 只要高严重度记录 curl localhost:9200/_ml/anomaly_detectors/cpu-spike/results/records?record_score90 # 最多 5 条按时间先后排序 curl localhost:9200/_ml/anomaly_detectors/cpu-spike/results/records?size5实现上get_ml_recordses_compat.rs先按record_score min保留再按timestamp升序显式排序最后truncate(size)截断返回{ count: N, records: [...] }。注意count是过滤/截断之后的条数。结果存储本身也有硬上限每个 job 最多保留ML_MAX_RECORDS_PER_JOB 10_000条超出后从最旧的一端淘汰es_compat.rs。6. 停止与删除curl -XPOST localhost:9200/_ml/datafeeds/cpu-spike-feed/_stop # - {stopped: true} (中止后台任务) curl localhost:9200/_cat/ml/datafeeds # cpu-spike-feed stopped cpu-spike curl -XDELETE localhost:9200/_ml/datafeeds/cpu-spike-feed # {acknowledged: true}_stop把状态置为stopped并abort()掉后台评分任务对不存在的 id 返回404datafeed 必须先停止才能删除——删除一个started状态的 datafeed 返回409status_exception与 Elasticsearch 行为一致delete_ml_datafeedes_compat.rs_cat/ml/datafeeds每个 datafeed 一行列固定为id state job_id也支持formatjson。复现一个纯标准库的端到端脚本两个终端、零外部依赖——脚本只用 Python 3 标准库不用pip install、没有第三方服务、没有 API key位于 docs/examples/continuous-anomaly-datafeeds/datafeed_demo.py# 1. 起一个一次性 XERJ默认监听 :9200 xerj --insecure # 2. 另一个终端里从仓库根目录运行演示 python3 docs/examples/continuous-anomaly-datafeeds/datafeed_demo.py # 指向其他 host/port 用 XERJ_URLXERJ 仍作为别名可用 XERJ_URLhttp://localhost:9200 python3 docs/examples/continuous-anomaly-datafeeds/datafeed_demo.py脚本流程与本文六节一一对应建索引 → 灌入含第一个尖峰的序列 → 创建检测器和 datafeed →_start并完整打印第一条异常记录 → 向实时索引注入第二个尖峰 → 轮询直到运行中的 datafeed 自己抓到它不二次_start→ 验证record_score90过滤与size1截断 → 停止、_cat确认、删除。最后它用一组assert校验两个尖峰都被检测到、第二个无需重新启动任何一环回归即以非零码退出——所以它同时是一个冒烟测试。指标序列是固定的所以数值逐次运行确定00:12 桶actual 96.0对expected 20.25std_dev 1.1636866703140785、z_score 65.09、record_score 100.0连续检测到的 00:20 桶actual 88.0对着同一基线。只有墙上时间会变约 2 秒一两个2s心跳。预期输出datafeed cpu-spike-feed - job cpu-spike (every 2s), statestopped raw datafeed config: { datafeed_id: cpu-spike-feed, frequency: 2s, indices: [ cpu_datafeed ], job_id: cpu-spike, query: { match_all: {} }, state: stopped } after _start — 1 anomaly record(s) from the first pass: when actual expected record_score 2026-07-06 00:12 UTC 96.0 20.2 100.0 raw first record: { actual: 96.0, anomaly_score: 100.0, bucket_span: 1m, expected: 20.25, is_anomaly: true, job_id: cpu-spike, record_score: 100.0, result_type: record, std_dev: 1.1636866703140785, timestamp: 1783296720000, timestamp_iso: 2026-07-06T00:12:00.000Z, typical: 20.25, z_score: 65.09484205018445 } injecting a new spike at 00:20, waiting for the 2s scorer... datafeed picked up 1 new anomaly record(s) on its own: when actual expected record_score 2026-07-06 00:20 UTC 88.0 20.2 100.0 filter record_score90 - 2 record(s); size1 cap - 1 record(s) _stop - {stopped: True} · datafeed state now: stopped _cat/ml/datafeeds - cpu-spike-feed stopped cpu-spike DELETE datafeed - {acknowledged: True} OK: 2 anomalies detected continuously; datafeed stopped cleanly.端点速查表方法路径用途PUT/_ml/datafeeds/{id}创建/替换 datafeedjob_id必填indices、query、frequency可选GET/_ml/datafeeds/{id}获取单个 datafeedGET/_ml/datafeeds列出所有 datafeed返回{datafeeds, count}按 id 排序DELETE/_ml/datafeeds/{id}删除 datafeed必须先停止否则409POST/_ml/datafeeds/{id}/_start启动立即评分一次 每frequency重评分POST/_ml/datafeeds/{id}/_stop停止并中止后台评分器GET/_ml/anomaly_detectors/{job}/results/records已存储的异常记录支持record_score/size过滤GET/_cat/ml/datafeeds每个 datafeed 一行id state job_id这些路由统一注册在 engine/crates/xerj-api/src/router.rs 的_ml continuous datafeeds区块。延伸阅读检测器模型与这些记录背后的评分数学滑动基线、z-score、阈值语义见 Detect a spike in a metric series数据结构定义MlDetector与MlDatafeed含持久化路径与重启重置逻辑在 engine/crates/xerj-api/src/state.rs全部_ml处理函数put_ml_datafeed、start_ml_datafeed、datafeed_score_pass、spawn_datafeed_task、get_ml_records在 engine/crates/xerj-api/src/es_compat.rsdatafeed 后台任务不越权读取受限索引的回归测试engine/crates/xerj-api/tests/ml_datafeed_cannot_read_a_brain.rs。赞分享【免费下载链接】xerjXERJ is the new way for AI to search data. Its autoindex capability activates agents to know your data without the token waste of grep and sed. One command indexes code, docs, logs and PDFs for search, RAG, security audits and agent memory, using 40x fewer tokens than grep. Elasticsearch compatible, so existing clients just work.项目地址https://gitcode.com/GitHub_Trending/xe/xerj点击查看免费下载相关推荐用 XERJ 检测指标序列中的尖峰异常_ml 异常检测配方与源码级解读用 XERJ 检测指标序列中的尖峰异常_ml 异常检测配方与源码级解读 XERJ 内置了一个按需on demand的异常检测器接口采用 ElasticsXERJ Postgres CDC 实战用逻辑复制让 PostgreSQL 数据自动同步进 XERJ并跑起 RRF 混合检索XERJ Postgres CDC 实战用逻辑复制让 PostgreSQL 数据自动同步进 XERJ并跑起 RRF 混合检索 本文以 daily.dev 的gh_mirrors/do/dockerfiles与监控指标告警案例异常检测实战gh_mirrors/do/dockerfiles与监控指标告警案例异常检测实战 你是否曾因Docker容器异常崩溃而导致服务中断是否在排查问题时缺乏有效的云原生DevOps上一篇ML-For-Beginners 时间序列预测实战使用 Support Vector RegressorSVR建模电力负荷下一篇Apache APISIX WebSocket 认证实战在握手阶段用 key-auth 拦截未授权连接创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
