Logpush 以尽可能快的速度批量投递日志，使 Cloudflare 能够近乎实时地提供信息。在最近的 Birthday Week 期间，我们将 Logpush 开放给所有自助服务计划，并让 Transformers 正式可用。
这些变更是 Cloudflare Observability 的八项重大更新之一，体现了我们让每位用户都能使用所有 Cloudflare 功能的承诺。本文将深入探讨 Transformers 如何在 Logpush 的导出管道中工作。可观测性团队使用 Logpush 将 Cloudflare 日志导出到安全信息和事件管理系统（SIEM）、分析平台以及对象存储。在日志到达目的地之前，团队可能需要移除健康检查、省略敏感字段，或为仪表盘和调查重塑记录。如今，这通常需要团队部署、扩展、监控和维护一个单独的处理服务。
例如，一个从多个 CDN 收集日志的团队可以添加一个静态的 provider 字段，在记录到达其分析平台之前对其进行规范化。团队还需要为导出、摄取和存储那些一到就丢弃的记录付费，因此在投递前进行过滤可以减少下游数据量和成本。
借助 Transformers，这一处理过程被移入 Logpush 导出工作流中。团队可以定义一条 SQL 查询来过滤记录、脱敏敏感数据，并通过添加、重命名、删除或嵌套字段来重塑输出。Logpush 在导出前应用该查询，因此每个目的地都能以所需的形式接收数据。我们通过在 Logpush 的高吞吐量导出管道中嵌入 SQL 转换引擎，同时保持性能和资源使用可预测，使这一工作流对客户来说变得简单。
Logpush Transformers 的工作原理
Transformer（深度学习中的注意力模型架构） 是为特定 Logpush 数据集编写的 SQL 查询。该查询引用该数据集中的字段，并定义 Logpush 应导出哪些记录和字段。Logpush 在导出前对每个批次应用该查询。不匹配 WHERE 子句的记录会被移除。对于剩余记录，SELECT 语句的输出将成为导出的记录。Transformers 可以更改每条记录的结构和内容，转换后的结果以 NDJSON 格式导出。
客户可以在 Cloudflare 仪表板中通过 Transformer Studio 创建、预览和管理 Transformers，也可以通过 API（应用程序编程接口） 以编程方式进行操作。保存 Transformer 后，客户可以将其附加到使用相同数据集的兼容 NDJSON Logpush 作业上。每次更新都会创建一个新版本，因此客户可以查看 Transformer 随时间的变化。
以下示例重塑了一条 Cloudflare HTTP 请求日志，以匹配团队在多个 CDN 提供商之间使用的自定义模式。它不会在序列化格式之间进行转换：输入和输出都使用 NDJSON。
{
"ClientIP"
:
"203.0.113.42"
,
"ClientRequestHost"
:
"example.com"
,
"ClientRequestMethod"
:
"POST"
,
"ClientRequestPath"
:
"/api/checkout"
,
"ClientRequestUserAgent"
:
"Mozilla/5.0"
,
"EdgeResponseStatus"
:
502
,
"EdgeStartTimestamp"
:
"2026-05-21T15:00:00Z"
,
"RayID"
:
"8e2a1c60ef9e1c9a"
}该 Transformer 重命名并嵌套字段，对客户端 IP 地址的一部分进行脱敏，添加 provider 和 error 信息，移除未使用的字段，并过滤掉内部流量和健康检查流量：
SELECT extract(epoch FROM to_timestamp(EdgeStartTimestamp)) AS unix_ts, regexp_replace(ClientIP, '\.\d+\.\d+$' , '.XXX.XXX' ) AS redacted_ip, UPPER (ClientRequestMethod) AS method, EdgeResponseStatus AS status , RayID AS ray_id, 'cloudflare' AS provider , named_struct( 'host' , ClientRequestHost, 'path' , ClientRequestPath ) AS request, EdgeResponseStatus >= 500 AS is_server_error FROM http_requests WHERE ClientRequestHost NOT IN ( 'internal.example.com' , 'health.example.com' ) AND ClientRequestPath NOT LIKE '/healthz%' AND ClientRequestPath NOT LIKE '/metrics%'
生成的 NDJSON 记录使用规范化模式：
{
"unix_ts"
:
1779375600
,
"redacted_ip"
:
"203.0.XXX.XXX"
,
"method"
:
"POST"
,
"status"
:
502
,
"ray_id"
:
"8e2a1c60ef9e1c9a"
,
"provider"
:
"cloudflare"
,
"request"
: {
"host"
:
"example.com"
,
"path"
:
"/api/checkout"
},
"is_server_error"
:
true
}Transformers 如何使用 Apache DataFusion 和 Arrow 执行 SQL
Cloudflare 的数据管道每秒摄取超过十亿个事件。其 Logpush 投递服务（我们称之为 Bufstream）每秒向客户目的地投递超过 1.1 亿条 Logpush 记录。它运行在数百个 Kubernetes（容器编排平台） 容器编排平台 Pod 上，每个 Pod 从 Buftee（Cloudflare 基于日志的分布式内部消息代理）中认领一个时间切片的数据。这种模式将工作横向分布到各个 Logpush 作业中。作业量遵循类帕累托分布：大多数作业处理量较低，而少数高流量作业每秒处理数百万条日志。Transformer（深度学习中的注意力模型架构）需要满足与 Logpush 其余部分相同的运维要求：可靠投递、高吞吐量，以及在不向导出热路径引入额外故障场景的前提下实现可预测的资源使用。
我们最初的原型在 Cloudflare Workers 中运行客户定义的 WebAssembly Transformer，Workers 提供全球分布式、隔离的计算能力。然而，调用 Worker 会在 Logpush 导出路径中插入一次网络调用。这种远程依赖可能引入瞬时超时和重试，从而降低投递吞吐量并增加投递延迟。
我们决定将 Transformer 执行保留在投递服务进程内，以优先保障可靠性和投递吞吐量。我们评估了多个 SQL 执行引擎，最终选择了 Apache DataFusion，它已经为 Cloudflare Basin 提供支持。
Apache DataFusion 是一个基于 Apache Arrow 构建的 Rust 查询引擎，具有 SQL 规划器和向量化、列式执行引擎。在一项基准测试中，从单个 Go goroutine 对合成的 5,000 行 http_requests 批次执行一个准备好的 Transformer 大约需要 3 毫秒，相当于每秒约 160 万行。
将 Transformer 附加到作业会为其处理路径增加 CPU 工作。Bufstream 使用按作业的转换并行度，因此高流量作业可以并发执行转换批次。将 DataFusion 嵌入 Bufstream 使这些批次能够使用该服务现有的投递基础设施，包括其调度、批处理和目的地控制，而无需引入单独的转换层或网络跳转。
选择 DataFusion 意味着需要将用 Go 编写的 Bufstream 与用 Rust 编写的引擎连接起来。下一个挑战是如何高效地跨过这一边界传输带类型的批次。
进程内 Transformer 管道的最终设计如下图所示。
从记录到列
Bufstream 从 Buftee 接收包含 Cap'n Proto 记录的压缩 Logpush 批次。对于附加了 Transformer 的作业，Bufstream 会解压批次、解析符合条件的日志，并在执行转换之前构建带类型的 Apache Arrow RecordBatch。
Apache Arrow 是一种与语言无关的列式内存格式，专为在现代 CPU 和 GPU（图形处理器） 图形处理器上高效分析而设计。它将扁平数据和嵌套数据表示为带类型的连续列。Arrow IPC 为 Bufstream 和 DataFusion 提供了一种标准方式，用于在 Go 和 Rust 之间的外部函数接口（FFI）边界上交换批次，该接口通过 C 应用程序二进制接口（ABI）暴露，并使用 cgo（Go 调用 C 代码的机制）从 Go 中调用。Go 和 Rust 运行时不会共享转换数据的所有权。Go 仅在 cgo 调用期间将其输入 Arrow IPC 缓冲区的指针传递给 Rust，Rust 在返回后不会保留该指针。Rust 拥有转换后的输出缓冲区；Go 将其复制到 Go 管理的内存中，然后显式释放 Rust 分配的内存。这使得跨语言边界的所有权和生命周期变得明确，避免了跨运行时的内存管理错误。
使用 Arrow IPC 意味着 Rust DataFusion 封装可以使用 Arrow StreamReader 读取带类型的 RecordBatch 值，而不是从 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 FROM http_requests WHERE EdgeResponseStatus >= 500
执行转换
当客户通过 API 或 Transformer Studio 保存 Transformer 时，Cloudflare 将其 SQL 查询存储在 R2 中。Bufstream 通过多层缓存检索该查询，并将其转换为可复用的 DataFusion 逻辑计划，该计划定义了扫描、过滤、投影和表达式。Bufstream 为作业数据的每个时间切片准备一次此计划，并在该切片中的每个 Arrow 批次中复用它。对于每个新的 Arrow 批次，Rust 封装器将该批次绑定为计划的表输入，执行查询，并返回包含所选记录和字段的 Arrow 批次。
Transformer 被设计为故障关闭（fail closed）。如果转换无法执行，Bufstream 既不会推进作业的投递偏移量，也不会发送未转换的批次。用原始数据代替转换后的数据可能会违反下游管道的 schema、过滤或脱敏假设，因此 Bufstream 选择重新尝试失败的转换。
为防止持续失败的作业消耗健康作业所需的容量，Bufstream 会逐步延迟重试并在作业上报告错误。长时间处于失败状态的作业会被自动禁用。客户可以在 Logpush 中 Health 标签页的 Transformer Health 视图中监控成功和失败的转换、转换前后的数据量以及转换持续时间。
回到导出管道
执行后，Bufstream 将 Arrow 输出转换为 NDJSON 日志。它将这些日志分批、gzip 压缩并上传到配置的目标位置。不同目标位置接受数据的速率不同，并且可能会暂时变慢。Bufstream 在遵守目标位置并发限制的同时调整上传 goroutine 的数量。
通过将 SQL 执行嵌入 Bufstream，Transformer 利用了 Logpush 现有的分布式基础设施、批处理和目标位置控制。Cloudflare 的数据管道处理这些复杂性，而客户只需几个步骤即可创建并应用 Transformer。
开始使用
从 Cloudflare 仪表板的 Logpush 页面打开 Transformer Studio，然后按照以下步骤操作：
选择要转换的 Logpush 数据集。
编写 SQL 查询并针对示例记录进行预览。
保存 Transformer。
将其附加到匹配的 Logpush 作业。
转载声明：本文为转载整理，原文版权归原作者及原站点所有。
原文来源：Cloudflare Blog
原文作者：Sahidya Devadoss；如有侵权，请联系管理员删除。
