メトリクスと分析

Pipelines はメトリクスを公開します。取り込んだデータ、処理したデータ、sink へ配信したデータを測れます。

Cloudflare ダッシュボード ↗ に表示されるメトリクスは、Cloudflare の GraphQL Analytics API から照会されます。メトリクスには GraphQL または HTTP クライアントからプログラムでアクセスできます。

メトリクス

オペレーターメトリクス

Pipelines は、次のメトリクスを pipelinesOperatorAdaptiveGroups データセットで公開します。これらのメトリクスは、パイプラインオペレーターが読み取り・処理したデータを追跡します。

メトリクス	GraphQL フィールド名	説明
受信バイト数	`bytesIn`	パイプラインが読み取ったバイトの合計（`streamId_neq: ""` でフィルタすると、ストリームから読み取ったデータを取得できます）
受信レコード数	`recordsIn`	パイプラインが読み取ったレコードの合計（`streamId_neq: ""` でフィルタすると、ストリームから読み取ったデータを取得できます）
デコードエラー数	`decodeErrors`	ストリームスキーマでデシリアライズできなかったメッセージ数

イベントがドロップされた理由（missing_field、type_mismatch、parse_failure、null_value などの具体的なエラータイプを含む）の内訳は、ユーザーエラーメトリクスを参照してください。

pipelinesOperatorAdaptiveGroups データセットでは、照会のフィルタとグループ化に次のディメンションを使えます。

pipelineId - パイプラインの ID
streamId - ソースストリームの ID
datetime - 操作のタイムスタンプ
date - 操作のタイムスタンプ（日の始まりに切り詰め）
datetimeHour - 操作のタイムスタンプ（時の始まりに切り詰め）

Sink メトリクス

Pipelines は、次のメトリクスを pipelinesSinkAdaptiveGroups データセットで公開します。これらのメトリクスは、sink へのデータ配信を追跡します。

メトリクス	GraphQL フィールド名	説明
書き込みバイト数	`bytesWritten`	圧縮後に sink へ書き込んだバイトの合計
書き込みレコード数	`recordsWritten`	sink へ書き込んだレコードの合計
書き込みファイル数	`filesWritten`	sink へ書き込んだファイル数
書き込み行グループ数	`rowGroupsWritten`	書き込んだ行グループ数（Parquet ファイル向け）
圧縮前書き込みバイト数	`uncompressedBytesWritten`	圧縮前に書き込んだバイトの合計

pipelinesSinkAdaptiveGroups データセットでは、照会のフィルタとグループ化に次のディメンションを使えます。

pipelineId - パイプラインの ID
sinkId - 宛先 sink の ID
datetime - 操作のタイムスタンプ
date - 操作のタイムスタンプ（日の始まりに切り詰め）
datetimeHour - 操作のタイムスタンプ（時の始まりに切り詰め）

ユーザーエラーメトリクス

Pipelines は、デシリアライズエラーにより処理中にドロップされたイベントを追跡します。構造化ストリームが定義済みスキーマに一致しないイベントを受け取ると、取り込み時には受け付けられますが、処理中にドロップされます。pipelinesUserErrorsAdaptiveGroups データセットは、これらのドロップされたイベントを可視化し、どのイベントがなぜドロップされたかを示します。このデータセットの完全なスキーマは、GraphQL のイントロスペクションで確認できます。

メトリクス	GraphQL フィールド名	説明
件数	`count`	検証に失敗したイベントの数

pipelinesUserErrorsAdaptiveGroups データセットでは、照会のフィルタとグループ化に次のディメンションを使えます。

pipelineId - パイプラインの ID
errorFamily - エラーのカテゴリ（例: deserialization）
errorType - ファミリー内の具体的なエラータイプ
date - エラーの日付（日の始まりに切り詰め）
datetime - エラーのタイムスタンプ
datetimeHour - エラーのタイムスタンプ（時の始まりに切り詰め）
datetimeMinute - エラーのタイムスタンプ（分の始まりに切り詰め）

既知のエラータイプ

エラーファミリー	エラータイプ	説明
`deserialization`	`missing_field`	ストリームスキーマで必須のフィールドがイベントに存在しなかった
`deserialization`	`type_mismatch`	フィールドの値がスキーマの想定型と一致しなかった（例: 数値が想定される箇所に文字列を送信）
`deserialization`	`parse_failure`	イベントを有効な JSON として解析できなかった、またはフィールド値を想定型へ解析できなかった
`deserialization`	`null_value`	必須フィールドは存在したが、値が null だった

ダッシュボードでメトリクスとエラーを確認する

パイプラインごとの分析は、Cloudflare ダッシュボードで確認できます。パイプラインの現在および過去のメトリクスを見る手順は次のとおりです。

Cloudflare ダッシュボード ↗ にログインし、アカウントを選択します。
Pipelines > Pipelines を開きます。
パイプラインを選択します。
メトリクスを見る場合は Metrics タブ、ドロップされたイベントを見る場合は Errors タブを開きます。

照会する時間枠は任意で選べます。初期値は直近 24 時間です。

GraphQL API で照会する

GraphQL Analytics API を使い、パイプラインの分析をプログラムから照会できます。この API は Cloudflare ダッシュボードと同じデータセットを照会し、GraphQL のイントロスペクションにも対応しています。

Pipelines の GraphQL データセットでは、Cloudflare アカウント ID を指定する accountTag フィルターが必要です。

期間内のオペレーターメトリクスを測る

このクエリは、パイプラインがストリームから読み取ったバイト数とレコード数の合計と、デコードエラーを返します。

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

ドロップされたイベントのエラーを照会する

このクエリは、スキーマ検証の失敗によりドロップされたイベントの概要を、エラータイプごとにまとめ、件数の多い順で返します。

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]
)

アカウント上のすべてのパイプラインのエラーを照会するには、pipelineId フィルターを外し、ディメンションに pipelineId を含めます。

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