指标和分析

Pipelines 公开指标，使您能够衡量摄取、处理并交付到 sink 的数据量。

Cloudflare 仪表板 ↗中显示的指标从 Cloudflare 的 GraphQL Analytics API 查询。您可以通过 GraphQL 或 HTTP 客户端以编程方式访问这些指标。

指标

运算符指标

Pipelines 在 pipelinesOperatorAdaptiveGroups 数据集中导出以下指标。这些指标跟踪 pipeline 运算符读取和处理的数据。

指标	GraphQL 字段名	描述
Bytes In	`bytesIn`	pipeline 读取的总字节数（按 `streamId_neq: ""` 过滤以获取从 stream 读取的数据）
Records In	`recordsIn`	pipeline 读取的总记录数（按 `streamId_neq: ""` 过滤以获取从 stream 读取的数据）
Decode Errors	`decodeErrors`	无法在 stream schema 中反序列化的消息数

有关事件被丢弃原因的详细分类（包括 missing_field、type_mismatch、parse_failure 和 null_value 等特定错误类型），请参阅用户错误指标。

pipelinesOperatorAdaptiveGroups 数据集提供以下维度用于过滤和分组查询：

pipelineId - pipeline 的 ID
streamId - 源 stream 的 ID
datetime - 操作的时间戳
date - 操作的时间戳，截断到一天的开始
datetimeHour - 操作的时间戳，截断到一小时的开始

Sink 指标

Pipelines 在 pipelinesSinkAdaptiveGroups 数据集中导出以下指标。这些指标跟踪交付到 sink 的数据。

指标	GraphQL 字段名	描述
Bytes Written	`bytesWritten`	写入 sink 的总字节数（压缩后）
Records Written	`recordsWritten`	写入 sink 的总记录数
Files Written	`filesWritten`	写入 sink 的文件数
Row Groups Written	`rowGroupsWritten`	写入的行组数（用于 Parquet 文件）
Uncompressed Bytes Written	`uncompressedBytesWritten`	压缩前写入的总字节数

pipelinesSinkAdaptiveGroups 数据集提供以下维度用于过滤和分组查询：

pipelineId - pipeline 的 ID
sinkId - 目标 sink 的 ID
datetime - 操作的时间戳
date - 操作的时间戳，截断到一天的开始
datetimeHour - 操作的时间戳，截断到一小时的开始

用户错误指标

Pipelines 跟踪处理期间因反序列化错误而丢弃的事件。当结构化 stream 收到与其定义的 schema 不匹配的事件时，这些事件在摄取期间会被接受但在处理期间会被丢弃。pipelinesUserErrorsAdaptiveGroups 数据集提供对这些丢弃事件的可见性，告诉您哪些事件被丢弃以及原因。您可以使用 GraphQL 内省探索此数据集的完整 schema。

指标	GraphQL 字段名	描述
Count	`count`	验证失败的事件数

pipelinesUserErrorsAdaptiveGroups 数据集提供以下维度用于过滤和分组查询：

pipelineId - pipeline 的 ID
errorFamily - 错误类别（例如 deserialization）
errorType - 类别内的特定错误类型
date - 错误日期，截断到一天的开始
datetime - 错误的时间戳
datetimeHour - 错误的时间戳，截断到一小时的开始
datetimeMinute - 错误的时间戳，截断到一分钟的开始

已知错误类型

错误系列	错误类型	描述
`deserialization`	`missing_field`	stream schema 中定义的必填字段在事件中不存在
`deserialization`	`type_mismatch`	字段值与 schema 中的预期类型不匹配（例如，在预期数字的位置发送了字符串）
`deserialization`	`parse_failure`	事件无法解析为有效的 JSON，或字段值无法解析为预期类型
`deserialization`	`null_value`	必填字段存在但值为 null

在仪表板中查看指标和错误

每个 pipeline 的分析数据可在 Cloudflare 仪表板中查看。要查看 pipeline 的当前和历史指标：

登录 Cloudflare 仪表板 ↗ 并选择您的账户。
前往 Pipelines > Pipelines。
选择一个 pipeline。
前往 Metrics（指标） 选项卡查看指标，或 Errors（错误） 选项卡查看丢弃的事件。

您可以选择性地选择查询的时间窗口。默认为最近 24 小时。

通过 GraphQL API 查询

您可以通过 GraphQL Analytics API 以编程方式查询 pipeline 的分析数据。此 API 查询与 Cloudflare 仪表板相同的数据集，并支持 GraphQL 内省。

Pipelines GraphQL 数据集需要带有 Cloudflare 账户 ID 的 accountTag 过滤器。

测量时间段内的运算符指标

此查询返回 pipeline 从 stream 读取的总字节数和记录数，以及任何解码错误。

query PipelineOperatorMetrics(
	$accountTag: String!
	$pipelineId: String!
	$datetimeStart: Time!
	$datetimeEnd: Time!
) {
	viewer {
		accounts(filter: { accountTag: $accountTag }) {
			pipelinesOperatorAdaptiveGroups(
				limit: 10000
				filter: {
					pipelineId: $pipelineId
					streamId_neq: ""
					datetime_geq: $datetimeStart
					datetime_leq: $datetimeEnd
				}
			) {
				sum {
					bytesIn
					recordsIn
					decodeErrors
				}
			}
		}
	}
}

测量 sink 交付指标

此查询返回写入特定 sink 的数据的详细指标，包括文件和压缩统计信息。

query PipelineSinkMetrics(
	$accountTag: String!
	$pipelineId: String!
	$sinkId: String!
	$datetimeStart: Time!
	$datetimeEnd: Time!
) {
	viewer {
		accounts(filter: { accountTag: $accountTag }) {
			pipelinesSinkAdaptiveGroups(
				limit: 10000
				filter: {
					pipelineId: $pipelineId
					sinkId: $sinkId
					datetime_geq: $datetimeStart
					datetime_leq: $datetimeEnd
				}
			) {
				sum {
					bytesWritten
					recordsWritten
					filesWritten
					rowGroupsWritten
					uncompressedBytesWritten
				}
			}
		}
	}
}

查询丢弃事件错误

此查询返回因 schema 验证失败而丢弃的事件摘要，按错误类型分组并按频率排序。

query GetPipelineUserErrors(
	$accountTag: String!
	$pipelineId: String!
	$datetimeStart: Time!
	$datetimeEnd: Time!
) {
	viewer {
		accounts(filter: { accountTag: $accountTag }) {
			pipelinesUserErrorsAdaptiveGroups(
				limit: 100
				filter: {
					pipelineId: $pipelineId
					datetime_geq: $datetimeStart
					datetime_leq: $datetimeEnd
				}
				orderBy: [count_DESC]
			) {
				count
				dimensions {
					date
					errorFamily
					errorType
				}
			}
		}
	}
}

响应示例：

{
	"data": {
		"viewer": {
			"accounts": [
				{
					"pipelinesUserErrorsAdaptiveGroups": [
						{
							"count": 679,
							"dimensions": {
								"date": "2026-02-19",
								"errorFamily": "deserialization",
								"errorType": "missing_field"
							}
						},
						{
							"count": 392,
							"dimensions": {
								"date": "2026-02-19",
								"errorFamily": "deserialization",
								"errorType": "type_mismatch"
							}
						},
						{
							"count": 363,
							"dimensions": {
								"date": "2026-02-19",
								"errorFamily": "deserialization",
								"errorType": "parse_failure"
							}
						},
						{
							"count": 44,
							"dimensions": {
								"date": "2026-02-19",
								"errorFamily": "deserialization",
								"errorType": "null_value"
							}
						}
					]
				}
			]
		}
	},
	"errors": null
}

您可以通过在过滤器中添加 errorType 来按特定错误类型过滤：

pipelinesUserErrorsAdaptiveGroups(
	limit: 100
	filter: {
		pipelineId: $pipelineId
		datetime_geq: $datetimeStart
		datetime_leq: $datetimeEnd
		errorType: "type_mismatch"
	}
	orderBy: [count_DESC]
)

要查询账户上所有 pipeline 的错误，省略 pipelineId 过滤器并在 dimensions 中包含 pipelineId：

pipelinesUserErrorsAdaptiveGroups(
	limit: 100
	filter: {
		datetime_geq: $datetimeStart
		datetime_leq: $datetimeEnd
	}
	orderBy: [count_DESC]
) {
	count
	dimensions {
		pipelineId
		errorFamily
		errorType
	}
}