定时流水线
在编辑器中搭建流水线
打开 流水线，选择 新建流水线。编辑器分为三部分。 设置。 为流水线命名，选择信号类型，并把 执行模式 设为 实时 或 定时。定时流水线还需要设置 运行间隔（两次运行之间的间隔）、首次运行往回读（第一次运行往回读多久）和 等待迟到数据（最新的事件留多长时间等它们到达，再由运行读取）。其下一行是 运行失败重试、投递失败重试、死信数据流 和 去重字段，参见定时与读取位点和重试与死信。 流程。 流程按事件经过的顺序展示流水线：
选择节点即可编辑。流程头部显示 校验通过、还有几处待填写，或阻止保存的问题：缺少来源或目标、步骤缺少名称或脚本，以及目标同时又是来源。第一个来源和第一个目标数据流会保存为流水线的
source_stream 与 target_stream，完整的列表保存为 function_steps 中的 sources 与 sinks。
试运行。 保存前让一条样例事件走完全部步骤，不会写入任何数据。编辑 样例事件，或从来源数据流取一条近期事件，然后选择 运行。有多个来源时，样例来自 指定样例来自哪个来源，步骤把它读作 %source_stream。结果显示每个步骤之后的事件，以及事件将写入的数据流或丢弃它的步骤。流水线运行时，在某一步失败的事件计为失败；配置了死信数据流时会写进去。
选择 创建流水线 保存。在已保存的流水线上，更多 提供运行历史、回填和删除。
步骤
处理步骤
添加步骤 提供以下几类步骤：
解析类、字段类和事件类步骤以表单方式填写，并以 VRL 执行。查看生成的 VRL 显示脚本，转为 VRL 编辑 把该步骤变成 自定义 VRL 步骤。
自定义 VRL 步骤是一段 VRL 脚本。从已保存的函数开始 可以用内置预设或任意已保存的函数填入脚本，之后仍可编辑。
内置预设随每个实例提供，且只读：
过滤与采样
过滤事件 和 采样 用 VRL 的abort 语句有意丢弃事件。被丢弃的事件不会写入任何目标，也不会进入死信数据流。运行时它计为丢弃，不算失败。
- 过滤事件 检查一个字段，由 匹配条件的事件 决定处理方式：保留，其余丢弃 或 丢弃。条件可以把值与文本或正则表达式比较、检查字段是否存在，或比较数字。数字条件匹配数字，或能读成数字的文本。
- 采样 按 保留比例（%） 保留事件，取值大于 0、不超过 100。设置 按字段一致采样（可选） 后，该字段取值相同的事件要么全部保留，要么全部丢弃，例如同一条链路的所有 Span。留空则逐条随机抽样。
%source_stream 读取事件来自哪个数据流。这个值不会成为事件的字段，除非步骤把它复制进去：
路由
内置的路由步骤决定每个事件写入哪个目标数据流，两种模式都支持。路由有两种形式：- 固定数据流；
- 按字段动态分流：写入由事件中某个字段指定的数据流。可以设置 流名前缀，字段缺失、不是字符串或不是有效流名的事件写入 兜底数据流。
_、- 与 .。.、.. 以及受保护的 _molesignal 这样的名称会被拒绝。不存在的数据流会在首次写入时创建。
在定时流水线中，路由只替换第一个目标数据流，其余目标数据流不受路由影响，收到全部事件。路由不能把事件写回流水线读取的数据流：固定数据流 或 兜底数据流 是来源时，保存会以 400 拒绝；事件的字段指向某个来源时，该事件计为失败并写入死信数据流。
扩展表
扩展表 是转换可以使用的“键 → 记录”查找表。打开 扩展表，创建一张表，添加行（一个键加若干命名字段），然后在 VRL 中做查找：实时流水线
实时流水线在接入期间、事件被存储之前处理每条日志事件。可以在编辑器中把 执行模式 设为 实时 来创建，也可以通过 API：appname 为 checkout 的事件会进入数据流 app_checkout，没有 appname 的事件进入兜底数据流 default。
规划时需要了解的行为：
- 每个事件按顺序经过各个步骤。某个步骤失败，或结果不是对象，会拒绝该事件。接入响应会列出它，例如
{"index":1,"reason":"realtime transformneeds-levelfailed"}，同一批中的其他事件仍会被接收。 - 同时保留到源数据流 会把原始事件与转换后的副本一起保留在来源数据流中，默认关闭。来源与目标是同一个数据流时，只写入转换后的事件。
- 步骤执行
abort会有意丢弃该事件：不做路由，接入响应把它计为已接收，不报错误。来源与目标不同时，同时保留到源数据流 仍会保留原始事件。 - 每个来源数据流只能启用一条实时流水线，再创建第二条会返回
409。实时流水线只支持日志。 - 配置变更对之后的批次生效，已经开始处理的批次继续使用它读取到的配置。
- 流水线的输出不会再触发另一条流水线。
- 向多个目标数据流的写入不是原子的。整批重试可能让已经成功的数据流出现重复事件，因此该流水线不承诺恰好一次投递。
- 实时流水线不产生运行记录，也不能回填。列表中的 等待接收 状态表示配置已启用，并不表示数据已经到达。
定时流水线
定时流水线按间隔读取它的来源数据流，应用转换链，并把输出写入它的目标数据流。它还可以把同一批事件投递到连接器。目标与连接器出口
定时流水线通过标准接入写入最多 8 个 目标数据流（写入后可被查询），并投递到外部 连接器：- 每条事件写入第一个目标数据流，即
target_stream，除非路由步骤为它选了别的数据流。 - 其余每个目标数据流都收到经过各步骤的全部事件。
- 每个已启用的连接器都收到经过各步骤的全部事件。
在 流水线 → 连接器 添加连接器，再用 添加目标 → 连接器 把它们加入流程的 目标。一条流水线在同一次运行里写入它的数据流并投递到它的连接器。流水线不能写入它读取的数据流：目标数据流同时是来源时，会以
409 拒绝。
目标数据流不必事先存在：第一次写入时按输出推断 schema 建流，之后的运行像其他数据流一样补充新字段。事件的字段类型与已有目标数据流的 schema 冲突时，只拒收该事件：它计为失败并写入死信数据流，同一次运行的其余事件照常写入。如果某个目标数据流拒收了一次运行的全部事件，而流水线没有死信数据流，这次运行会失败，原因出现在它的 error 中。回填的写入方式相同。
把来源数据流从查询中隐藏
当一个兜底来源数据流扇出成每服务或每租户的目标流时，来源数据流会继续堆积原始事件。要让查询和仪表盘只对准路由后的数据流，把来源数据流标记为 不可查询：打开 数据流 → 来源 → 设置，关闭 可查询。 不可查询的数据流照常接收并保留数据，也照常为流水线供数。它会从查询选择器中隐藏，SQL 与 PromQL 搜索会以stream is not queryable 拒绝它。随时可以再打开开关。
定时与读取位点
Runner 不解析标准 cron 表达式。不符合
every: 语法的计划会被拒绝并返回 400：scheduled pipelines require every:Ns/m/h and a positive lookback。
流水线为每个来源保存一个读取位点：该来源在这个时间之前的事件都已处理。每次运行从各来源的读取位点读到「运行时间减去等待迟到数据的时长」，应用转换链并写出结果。读取位点随后移到这次读到的地方，下一次运行从那里接着读。在等待时长内到达的事件只读取一次。更晚到达、时间戳已被运行越过的事件不会被读取，因此来源有迟到数据时，应调大 等待迟到数据。
- 每个来源第一次运行往回读
lookback_secs。之后新加的来源也一样。 - 每次运行对每个来源最多读取运行间隔、
lookback_secs与一小时三者中最长的一段。落后更多的来源会在下一次检查时再次运行，一段一段地追赶，直到赶上。 - 重新启用已暂停的流水线会从头开始：第一次运行从运行时刻往回读
lookback_secs，不管暂停前读到了哪里。这个窗口之外的时间可以提交回填来处理。
定时 runner 在
alert 角色上运行（standalone 包含该角色）。存在多个 alert 节点时，每条流水线归一个节点负责，因此启用的流水线每个间隔只触发一次，而不是每个节点一次，参见分角色部署。重试与死信
读取某个来源或写入某个数据流失败时（例如存储不可用），该来源的读取位点保持不动：
- 同一个窗口在 30 秒后重试，之后每次等待时间加倍，最长不超过运行间隔。
- 重试只写入之前没有收下这个窗口的目标。
- 重试次数用完后跳过这个窗口：读取位点越过它，运行的
warning写明是哪个来源、哪一段时间没有处理。 - 每次尝试都会遇到的失败（例如配置无效）会让读取位点停在原处，不安排重试。每次常规运行都从同一位置重新读取，因此修复流水线期间不会跳过任何数据。
transform 死信保留事件读取时的样子，write 与 deliver 死信保留事件发送时的样子。没有死信数据流时，这些事件计为失败，不会保留。
去重字段 在各步骤之后生效，范围是同一次运行内的全部来源：保留该字段每个值的第一条事件，之后值相同的事件被丢弃。没有该字段的事件全部保留。事件中没有同名的完整字段时，user.id 这样的名称会访问嵌套对象。
运行历史
每次读取了窗口的运行都会记录。可以在流水线的 运行历史 标签页、编辑器的 更多 → 运行历史 中查看，也可以用GET /api/v1/scheduled_pipelines/{id}/runs。
运行读入的每条事件最终落入其中一项：
- 写出：它应去的每个目标都收下了它。
- 丢弃：某个步骤用
abort丢弃了它，或 去重字段 去掉了它。 - 失败：某个步骤处理它时失败、它被路由到了来源，或某个目标没有收下它。
回填
要按需处理历史窗口，在编辑器中打开 更多 → 回填，设置 起始 和 结束，然后选择 提交回填。也可以通过 API 提交：202，每个来源一个任务，各带 job_id 与 monitor 路径。每个任务通过异步搜索任务 worker 运行：在窗口内读取它的来源，应用同一条转换链，并像运行一样写入、投递和保存死信。
回填不改变读取位点，因此定时运行照常读取原本要读的数据。回填窗口与运行已经处理过的时间重叠时，这些事件会再写一遍。
权限
pipelines.read：列出流水线并读取运行历史。pipelines.create：创建流水线。pipelines.edit：修改流程、步骤、调度或回看窗口。pipelines.pause：启用或暂停流水线。pipelines.run：提交回填任务。pipelines.delete：删除流水线。
流水线 API
通过 HTTP API 创建、更新、列出与回填流水线。