Cloudflare Logpush 推出 Transformers：导出前用 SQL 过滤、脱敏和重塑日志

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 和错误信息、移除未使用的字段，并过滤掉内部流量和健康检查流量：

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）暴露，并在 Go 中通过 cgo（Go 调用 C 代码的机制）进行调用。

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 被设计为故障关闭。如果无法执行转换，Bufstream 既不会推进作业的交付偏移量，也不会发送未转换的批次。交付原始数据代替转换后的数据可能会违反下游管道的 schema、过滤或脱敏假设，因此 Bufstream 选择重新尝试失败的转换。

为防止持续失败的作业占用健康作业所需的容量，Bufstream 会逐步延迟重试并报告作业错误。长时间处于失败状态的作业会被自动禁用。客户可以在 Logpush 中 Health 标签页的 Transformer Health 视图里监控成功和失败的转换、转换前后的数据量以及转换时长。

返回导出流水线

执行完成后，Bufstream 将 Arrow 输出转换为 NDJSON 日志。它会对这些日志进行批处理、gzip 压缩，并上传到配置的目标位置。不同目标位置接受数据的速率不同，且可能暂时变慢。Bufstream 会在遵守目标并发限制的前提下调整上传 goroutine 的数量。

通过将 SQL 执行嵌入 Bufstream，Transformers 利用了 Logpush 现有的分布式基础设施、批处理和目标控制能力。Cloudflare 的数据流水线负责处理这些复杂性，而客户只需几个步骤即可创建并应用 Transformer。

开始使用

在 Cloudflare 仪表板中从 Logpush 页面 打开 Transformer Studio，然后按以下步骤操作：

1. 选择要转换的 Logpush 数据集。
2. 编写 SQL 查询，并针对示例记录进行预览。
3. 保存 Transformer。
4. 将其附加到匹配的 Logpush 作业。

Transformers 在 Free、Pro、Business 和 Enterprise 计划中均可用。有关设置详情，请参阅 Transformers 文档 ；有关当前定价信息，请参阅 Logpush 定价文档 。