← Cloudflare Pipelines / pipelines / observability

Metriky a analytika

Pipelines zpřístupňuje metriky, pomocí kterých můžete měřit data přijatá, zpracovaná a doručená do sinků.

Metriky zobrazené v Cloudflare dashboard ↗ jsou dotazovány z Cloudflare GraphQL Analytics API. K metrikám máte přístup přes programově přes GraphQL nebo HTTP klienta.

Metriky

Metriky operátoru

Pipelines exportuje níže uvedené metriky v rámci pipelinesOperatorAdaptiveGroups dataset. Tyto metriky sledují data načtená a zpracovaná operátory pipeline.

Metrika	Název pole GraphQL	Popis
Přijaté bajty	`bytesIn`	Celkový počet bajtů přečtených pipeline (filtrujte podle `streamId_neq: ""` pro čtení dat ze streamů)
Records In	`recordsIn`	Celkový počet záznamů přečtených pipeline (filtrujte podle `streamId_neq: ""` pro čtení dat ze streamů)
Decode Errors	`decodeErrors`	Počet zpráv, které se nepodařilo deserializovat podle schématu streamu

Podrobný přehled důvodů, proč byly události zahozeny (včetně konkrétních typů chyb, jako je missing_field, type_mismatch, parse_failure, a null_value), viz Metriky uživatelských chyb.

pipelinesOperatorAdaptiveGroups poskytuje tyto dimenze pro filtrování a seskupování dotazů:

pipelineId - ID pipeline
streamId - ID zdrojového streamu
datetime - časová značka operace
date - časová značka operace zaokrouhlená na začátek dne
datetimeHour - časová značka operace zaokrouhlená na začátek hodiny

Metriky sinku

Pipelines exportuje níže uvedené metriky v rámci pipelinesSinkAdaptiveGroups dataset. Tyto metriky sledují doručování dat do sinků.

Metrika	Název pole GraphQL	Popis
Zapsané bajty	`bytesWritten`	Celkový počet bajtů zapsaných do sinku po kompresi
Records Written	`recordsWritten`	Celkový počet záznamů zapsaných do sinku
Zapsané soubory	`filesWritten`	Počet souborů zapsaných do sinku
Zapsané skupiny řádků	`rowGroupsWritten`	Počet zapsaných row groups (u souborů Parquet)
Uncompressed Bytes Written	`uncompressedBytesWritten`	Celkový počet bajtů zapsaných před kompresí

pipelinesSinkAdaptiveGroups poskytuje tyto dimenze pro filtrování a seskupování dotazů:

pipelineId - ID pipeline
sinkId - ID cílového sinku
datetime - časová značka operace
date - časová značka operace zaokrouhlená na začátek dne
datetimeHour - časová značka operace zaokrouhlená na začátek hodiny

Metriky uživatelských chyb

Pipelines sleduje události, které se během zpracování zahodí kvůli chybám deserializace. Když strukturovaný stream přijme událost neodpovídající definovanému schématu, je při příjmu přijata, ale během zpracování zahozena. Metrika pipelinesUserErrorsAdaptiveGroups dataset poskytuje přehled o těchto zahozených událostech a ukazuje, které události byly zahozeny a proč. Celé schéma tohoto datasetu můžete prozkoumat pomocí GraphQL introspekce.

Metrika	Název pole GraphQL	Popis
Počet	`count`	Počet událostí, které neprošly validací

pipelinesUserErrorsAdaptiveGroups poskytuje tyto dimenze pro filtrování a seskupování dotazů:

pipelineId - ID pipeline
errorFamily - kategorie chyby (například deserialization)
errorType - konkrétní typ chyby v rámci rodiny
date - datum chyby zaokrouhlené na začátek dne
datetime - časová značka chyby
datetimeHour - časová značka chyby zaokrouhlená na začátek hodiny
datetimeMinute - časová značka chyby zaokrouhlená na začátek minuty

Známé typy chyb

Rodina chyb	Typ chyby	Popis
`deserialization`	`missing_field`	Povinné pole definované ve schématu streamu nebylo v události přítomno
`deserialization`	`type_mismatch`	Hodnota pole neodpovídala očekávanému typu ve schématu (například byl odeslán řetězec tam, kde se očekávalo číslo)
`deserialization`	`parse_failure`	Událost se nepodařilo zpracovat jako platný JSON, nebo se hodnotu pole nepodařilo převést na očekávaný typ
`deserialization`	`null_value`	Povinné pole bylo přítomno, ale mělo hodnotu null

Zobrazení metrik a chyb v dashboardu

Analytika jednotlivých pipeline je dostupná v Cloudflare dashboardu. Aktuální i historické metriky pipeline zobrazíte takto:

Přihlaste se do Cloudflare dashboard ↗ a vyberte svůj účet.
Přejděte na Pipelines > Pipelines.
Vyberte pipeline.
Přejděte na Metriky kartu pro zobrazení jejích metrik nebo Chyby kartu pro zobrazení zahozených událostí.

Volitelně můžete vybrat časové okno pro dotaz. Výchozí hodnota je posledních 24 hodin.

Dotazování přes GraphQL API

Analytická data pro své pipelines můžete programově dotazovat přes GraphQL Analytics API. Toto API dotazuje stejné datové sady jako Cloudflare dashboard a podporuje GraphQL introspekce.

Datové sady GraphQL pro Pipelines vyžadují accountTag filtr s ID vašeho účtu Cloudflare.

Měření metrik operátoru v čase

Tento dotaz vrátí celkový počet bajtů a záznamů, které pipeline přečetl ze streamů, spolu s případnými chybami dekódování.

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
				}
			}
		}
	}
}

Měření metrik doručování dat do sinku

Tento dotaz vrátí podrobné metriky o datech zapsaných do konkrétního sinku, včetně statistik souborů a komprese.

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
				}
			}
		}
	}
}

Dotazování na chyby zahozených událostí

Tento dotaz vrátí přehled událostí, které byly zahozeny kvůli chybám validace schématu, seskupených podle typu chyby a seřazených podle četnosti.

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
				}
			}
		}
	}
}

Příklad odpovědi:

{
	"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
}

Podle konkrétního typu chyby můžete filtrovat přidáním errorType do filtru:

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

Chcete-li dotazovat chyby napříč všemi pipelines v účtu, vynechte pipelineId filtr a zahrňte pipelineId v dimenzích:

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