导语：Cloudflare 为 Logpush 引入 Transformers，让企业在日志送达第三方系统之前完成过滤、字段重塑与敏感信息脱敏。
摘要
Cloudflare 为 Logpush 引入 Transformers，让企业在日志送达第三方系统之前完成过滤、字段重塑与敏感信息脱敏。此前这类处理通常需要另建一套加工服务，由其承担部署、扩容、监控与运维；同时，团队还要为那些一送到就被丢弃的记录支付导出、摄入与存储成本。Transformers 把这段加工移进 Logpush 的导出流程：用户写一条 SQL 查询来筛选记录、脱敏并重排字段结构，Logpush 在导出前执行。该功能已在 Free、Pro、Business、Enterprise 各档位开放。
正文（译文）
Logpush 以批处理方式尽快投递日志，使 Cloudflare 能够提供近乎实时的信息。在我们最近的一次更新中，我们把 Logpush 开放给了所有自助套餐，并为 Cloudflare Observability 带来了八项重要更新，这也体现了我们让每一个 Cloudflare 功能对所有人可用的承诺。本文进一步拆解 Transformers 在 Logpush 导出管线里的工作方式。
可观测性团队使用 Logpush，把 Cloudflare 日志导出到安全信息与事件管理系统（SIEM）、分析平台和对象存储。在这些日志到达目的地之前，团队可能需要剔除健康检查流量、省略敏感字段，或者为了让记录适配仪表盘与排查场景而重塑结构。今天，这通常需要一套独立的处理服务，团队必须自己部署、扩容、监控和维护。
举例来说，一个从多家 CDN 收集日志的团队，可以在记录送达分析平台之前，加上一个静态的 provider 字段来做归一化。此外，团队还要为那些一到就被丢弃的记录支付导出、摄入和存储的费用，因此在投递前先过滤，可以减少下游的数据量与成本。
有了 Transformers，这段加工被移进了 Logpush 的导出工作流。团队定义一条 SQL 查询，用来过滤记录、脱敏敏感数据，并通过新增、重命名、删除或嵌套字段来重塑输出。Logpush 在导出前应用这条查询，于是每个目的地收到的都是它所需要形态的数据。为了让这个工作流对用户足够简单，我们把一个 SQL 转换引擎嵌入了 Logpush 的高吞吐导出管线，同时保持性能与资源占用可预期。
Logpush Transformers 如何工作
Transformer 是一条针对特定 Logpush 数据集编写的 SQL 查询。查询引用该数据集中的字段，并定义 Logpush 应该导出哪些记录和哪些字段。Logpush 在导出前对每个批次应用这条查询：不匹配 WHERE 子句的记录被移除；对于剩下的记录，SELECT 语句的输出就成为被导出的记录。Transformers 可以改变每条记录的结构与内容，转换后的结果以 NDJSON 格式导出。
客户可以在 Cloudflare 仪表盘的 Transformer Studio 里创建、预览和管理 Transformers，也可以通过 API 以编程方式完成。保存一个 Transformer 之后，可以把它挂到使用同一数据集、兼容 NDJSON 的 Logpush 任务上。每次更新都会生成一个新版本，因此客户可以回看一个 Transformer 随时间的变化。
下面的例子把一条 Cloudflare HTTP 请求日志重塑为团队跨 CDN 厂商统一定义的 schema。它不做序列化格式之间的转换：输入与输出都是 NDJSON。
输入字段包括 ClientRequestUserAgent 等。
这条 Transformer 重命名并嵌套字段，对客户端 IP 地址做部分脱敏，补充 provider 与错误信息，去掉不用的字段，并过滤掉内部流量与健康检查流量：
SELECT to_timestamp(EdgeStartTimestamp)) ... regexp_replace(ClientIP, ...
（完整 SQL 示例见原文）
转换后得到的 NDJSON 记录即采用归一化后的 schema。
Transformers 如何用 Apache DataFusion 与 Arrow 执行 SQL
Cloudflare 的数据管线每秒摄入超过 10 亿个事件。它的 Logpush 投递服务（我们称之为 Bufstream）每秒向客户目的地投递超过 1.1 亿条 Logpush 记录。它运行在数百个 Kubernetes pod 上，每个 pod 从 Buftee（Cloudflare 自研的分布式、基于日志的消息代理）认领一个数据时间片。这种模型把工作横向分布到各个 Logpush 任务上。任务量呈帕累托式分布：多数任务量较小，少数高量任务每秒处理数百万条日志。Transformers 必须满足与 Logpush 其余部分相同的运维要求：可靠投递、高吞吐、资源占用可预期，同时不往导出热路径里引入新的故障场景。
我们最初的原型是在 Cloudflare Workers 里运行用户定义的 WebAssembly Transformers，它提供全球分布、隔离的计算环境。但调用一个 Worker 会在 Logpush 的导出路径里插入一次网络调用。这个远程依赖可能引入瞬时超时与重试，降低投递吞吐、增加投递时延。
我们决定把 Transformer 的执行保留在投递服务进程内，以优先保证可靠性与投递吞吐。我们评估了多个 SQL 执行引擎，最终选择了 Apache DataFusion――一个构建在 Apache Arrow 之上的 Rust 查询引擎，带有 SQL 规划器与向量化列式执行引擎。在一项基准中，从单个 Go goroutine 对一批合成的 5,000 行 http_requests 执行一个预编译 Transformer，耗时约 3 毫秒，相当于大约每秒 160 万行。
给任务挂上 Transformer 会给它的处理路径增加 CPU 工作。Bufstream 采用按任务的转换并行度，让高量任务可以并发执行转换批次。把 DataFusion 嵌入 Bufstream，使这些批次能够复用服务既有的投递基础设施，包括调度、批处理与目的地控制，而无需引入独立的转换层或网络跳转。
选择 DataFusion 意味着要把用 Go 写的 Bufstream 与用 Rust 写的引擎连接起来。下一个挑战是如何高效地让带类型的批次跨过这道边界。
Bufstream 从 Buftee 接收含 Cap'n Proto 记录的压缩 Logpush 批次。对于挂了 Transformer 的任务，Bufstream 解压缩批次、解析符合条件的日志，并在执行转换前构建带类型的 Apache Arrow RecordBatch。
Apache Arrow 是一种与语言无关的列式内存格式，面向现代 CPU 与 GPU 上的高效分析而设计。它把扁平与嵌套数据表示为带类型的连续列。Arrow IPC 为 Bufstream 与 DataFusion 提供了一种跨 Go 与 Rust 之间的外部函数接口（FFI）边界交换批次的标准方式，该边界通过 C 应用二进制接口（ABI）暴露，并由 Go 使用 cgo（Go 调用 C 代码的机制）调用。
Go 与 Rust 运行时并不共享转换数据的所有权。Go 仅在 cgo 调用期间把指向其输入 Arrow IPC 缓冲区的指针传给 Rust，Rust 在返回后不保留该指针。Rust 拥有转换后的输出缓冲区；Go 把它复制到 Go 管理的内存中，然后显式释放 Rust 的分配。这让跨语言边界的所有权与生命周期变得明确，避免了跨运行时的内存管理缺陷。
使用 Arrow IPC 意味着 Rust 的 DataFusion 封装可以直接使用 Arrow 的值，而不必从 JSON 这类中间格式逐行解析再重建带类型的 Arrow 列。字段类型、可空性与嵌套值在边界两侧一致表示，因此 DataFusion 拿到的是其原生列式模型的数据。
跨越 Go/Rust 边界并非零成本：批次仍需序列化、解码并在内存中管理。Arrow IPC 让这些工作保持批量导向，避免逐条记录的解析与转换。这使该边界更契合 Logpush 的高吞吐投递管线，同时保留针对 Logpush 数据集定义的 schema 校验。
每个 Arrow 批次都需要一个描述其字段与类型的 schema。Bufstream 不从某一批次里的日志去推断 schema，而是提供由每个 Logpush 数据集规格生成的 schema。这让 DataFusion 对可用字段与类型有稳定视图，从而能够做查询校验。Transformer 的 FROM 子句里指定的表名，决定查询使用哪个数据集 schema。例如这条 Transformer 查询的是 http_requests 数据集：
SELECT ClientIP, EdgeResponseStatus, RayID ...
当客户通过 API 或 Transformer Studio 保存一个 Transformer 时，Cloudflare 会把它的 SQL 查询存进 R2。Bufstream 通过多层缓存取回查询，并把它翻译成一个可复用的 DataFusion 逻辑计划，该计划定义扫描、过滤、投影与表达式。Bufstream 为每个任务数据的一个时间片准备一次该计划，并在该时间片内的每个 Arrow 批次上复用。对每个新的 Arrow 批次，Rust 封装把该批次绑定为计划的表输入，执行查询，并返回包含所选记录与字段的 Arrow 批次。
Transformers 的设计是"失败即闭合"的。如果某次转换无法执行，Bufstream 既不推进该任务的投递位点，也不发送未经转换的批次。用原始数据顶替转换后的数据，可能破坏下游管线的 schema、过滤或脱敏假设，因此 Bufstream 选择重试失败的转换。
为防止持续失败的任务占用健康任务所需的容量，Bufstream 会逐步延长重试间隔，并在任务上报错。长时间处于失败状态的任务会被自动停用。客户可以在 Logpush 的 Health 标签页下的 Transformer Health 视图中，监控转换成功与失败情况、转换前后的数据量，以及转换耗时。
回到导出管线
执行完成后，Bufstream 把 Arrow 输出转换为 NDJSON 日志，进行批处理、gzip 压缩，并上传到配置好的目的地。不同目的地的接收速率不同，也可能临时变慢。Bufstream 会在遵守目的地并发限制的前提下，调整上传 goroutine 的数量。
通过把 SQL 执行嵌入 Bufstream，Transformers 复用了 Logpush 既有的分布式基础设施、批处理与目的地控制。复杂性由 Cloudflare 的数据管线承担，而客户只需几步就能创建并应用一个 Transformer。
在 Cloudflare 仪表盘中打开 Transformer Studio，按以下步骤操作：
1 选择要转换的 Logpush 数据集。 2 编写你的 SQL 查询，并对一条样本记录做预览。 3 把它挂到匹配的 Logpush 任务上。
Transformers 已在 Free、Pro、Business 和 Enterprise 套餐上提供。设置细节请参阅 Transformers 文档，当前定价信息请参阅 Logpush 定价文档。
（来源：艾瑞网）
