Transformers

Transformers を使うと、Logpush が宛先へ配信する前に、各レコードバッチに対して SQL クエリを実行できます。保存したくないレコードの除外、下流スキーマに合わせたフィールドの整形、機密値のリダクション、新しいフィールドの計算、静的なメタデータの追加に使えます。

ロジックは 1 つの SQL クエリとして書き、Logpush ジョブに紐づけます。Cloudflare はすべてのバッチでそのクエリを実行します。FROM 句には Logpush データセット名（例: http_requests や audit_logs_v2）を指定し、フィールド名はそのデータセットのスキーマに従います。

• SQL ベースの変換 - Logpush ジョブごとに 1 文の SQL クエリです。
• レコード単位の実行 - Cloudflare がバッチを配信する前に、各 NDJSON レコードに対して実行します。
• 高度なフィルタリングと整形 - フィールドの削除、名前変更、リダクション、計算、タグ付けができます。
• 再デプロイなしで接続・解除 - Cloudflare ダッシュボードまたは API から Transformers を管理できます。
• バージョン履歴 - 保存するたびに新しいバージョンが作られ、古いバージョンも閲覧できます。

始める前に、次が必要です。

• 出力形式が ndjson の Logpush ジョブ。Transformers は NDJSON ジョブでのみ利用でき、アカウントスコープとゾーンスコープのどちらのデータセットでもサポートされます。
• アカウントに対する Logs Write 権限を持つ API トークン。
• 変換するレコードの データセット に関する知識。SQL はそのフィールド名を直接参照します。

Transformers の作成、プレビュー、接続、管理は、Cloudflare ダッシュボードまたは API から行えます。

Transformer Studio は、SQL エディターを含むワークスペースです。Transformers の作成、プレビュー、管理をここで行います。Cloudflare ダッシュボードの Logpush ページから開きます。

Studio では次ができます。

• Logpush データセットに対する SQL クエリを書いて、新しい transformer を作成する。
• 保存前に、対象データセット用の用意されたサンプルレコードで transformer をプレビューする。
• アカウントまたはゾーン上の、対象となる任意の Logpush ジョブに transformer を接続する。接続先として表示されるのは、transformer のデータセットに一致する NDJSON ジョブだけです。CSV ジョブは表示されません。
• ジョブから transformer を解除する。
• SQL を編集して保存するたびに、新しいバージョンを保存する。古いバージョンは閲覧できます。
• transformer の名前や説明を更新する。
• transformer を削除する。いずれかの Logpush ジョブが参照しているあいだは、transformer を削除できません。

transformer の操作はすべて Cloudflare API から行えます。認証には、Logs Write 権限を持つ API トークン を使います。

ジョブへの接続または解除は、Logpush ジョブの transformer_id を設定します。ジョブのエンドポイントは Logpush ジョブのセットアップ を参照してください。

transformer は 1 つの SQL クエリです。Logpush データセットがソーステーブルになり、クエリの出力が配信されるレコードになります。

SELECT ClientIP, RayID, EdgeResponseStatus
FROM http_requests
WHERE EdgeResponseStatus >= 400

FROM のテーブル名は、transformer を接続する Logpush ジョブのデータセットと一致している必要があります。一致しない場合、接続は失敗します。

WHERE 句に一致しないレコードは、出力から除外されます。

Transformers は Cloudflare Pipelines と同じ SQL 方言を使います。次の操作をサポートします。

• 射影 - 特定のフィールドを SELECT し、AS で名前を変更し、式で新しいフィールドを計算します。
• フィルタリング - 標準の比較、ブール、NULL チェック演算子を使った WHERE 句です。
• CTE - WITH ... AS (...) の共通テーブル式です。
• UNNEST - 配列またはリストのフィールドを行に展開します。
• JSON アクセス - -> 演算子は JSON オブジェクトを返し、->> は文字列を返します。例: RequestHeaders ->> 'Host'。
• 入れ子の出力 - named_struct('key', value, ...) は入れ子の JSON オブジェクトを組み立てます。
• 配列の出力 - [value1, value2] は JSON 配列を組み立てます。
• スカラー関数 - UPPER、LOWER、COALESCE、CAST、extract、to_timestamp など、標準的な SQL 関数です。

• 結合
• サブクエリ
• 集約（GROUP BY、HAVING、COUNT、SUM）
• ウィンドウ関数
• ORDER BY
• 複数ステートメント - transformer あたり 1 クエリです

すべての SQL クエリは、保存前に対象データセットのスキーマに対して検証されます。未知のフィールド、型の不一致、無効な構文、未知のデータセット、サポートされない操作は、アップロード時に拒否されます。

ダッシュボードでは、検証エラーがエディター内に行番号と列番号付きで表示されます。API では、レスポンスの errors 配列で返されます。

次の例は、主な機能 の各能力を 1 つのクエリにまとめたものです。いずれも一部のレコードだけを残し、残ったレコードを整形し、下流パイプラインが不要なフィールドを落とします。

この transformer は、監査証跡のうち update アクションだけを残し、残ったレコードを下流配信向けに整形します。具体的には次を行います。

• ActionType が update でないレコードをすべて除外します。
• ActionTimestamp を RFC3339 から Unix epoch の整数へ変換し、unix_ts に名前を変更します。
• ActionType を大文字にし、action_type に名前を変更します。
• 値 Cloudflare の固定フィールド provider を追加します。
• ActorType、ActorEmail、ActorIPAddress を入れ子の actor オブジェクトにまとめます。
• ResourceType = 'zone' からブール値の is_zone フラグを導出します。
• ResourceType と ResourceID から resource_meta 配列を組み立てます。
• SELECT から省くことで、ActorID、AccountID、ActorContext を落とします。

audit_logs_v2 データセットの入力レコード:

{
  "ActionType": "update",
  "ActorEmail": "user@example.com",
  "ActorID": "a1b2c3d4",
  "ActorIPAddress": "203.0.113.42",
  "ActorType": "user",
  "ActionTimestamp": "2026-05-21T15:00:00Z",
  "AccountID": "90796717",
  "ResourceID": "r1s2t3u4",
  "ResourceType": "zone",
  "ActorContext": "dashboard"
}

Transformer:

SELECT
  extract(epoch FROM to_timestamp(ActionTimestamp)) AS unix_ts,
  UPPER(ActionType) AS action_type,
  'Cloudflare' AS provider,
  named_struct(
    'type', ActorType,
    'email', ActorEmail,
    'ip', ActorIPAddress
  ) AS actor,
  ResourceType = 'zone' AS is_zone,
  [ResourceType, ResourceID] AS resource_meta
FROM audit_logs_v2
WHERE ActionType = 'update'

配信されるレコード:

{
  "action_type": "UPDATE",
  "actor": {
    "email": "user@example.com",
    "ip": "203.0.113.42",
    "type": "user"
  },
  "is_zone": true,
  "provider": "Cloudflare",
  "resource_meta": ["zone", "r1s2t3u4"],
  "unix_ts": 1779375600
}

この transformer は、ヘルスチェック、メトリクススクレイパー、内部向けホスト名を除くすべての HTTP トラフィックを残します。ログストリーム全体は欲しいが、予測できるノイズは除きたいチーム向けの一般的なパターンです。具体的には次を行います。

• internal.example.com と health.example.com へのリクエストをすべて除外します。
• /healthz または /metrics で始まるパスへのリクエストをすべて除外します。
• EdgeStartTimestamp を RFC3339 から Unix epoch の整数へ変換し、unix_ts に名前を変更します。
• ClientRequestMethod を大文字にし、method に名前を変更します。
• 値 Cloudflare の固定フィールド provider を追加します。
• ClientRequestHost、ClientRequestPath、ClientRequestMethod を入れ子の request オブジェクトにまとめます。
• EdgeResponseStatus >= 500 からブール値の is_server_error フラグを導出します。
• ClientRequestHost と ClientRequestPath から request_meta 配列を組み立てます。
• SELECT から省くことで、ClientIP、ClientRequestUserAgent、RayID、OriginResponseTime、WAFAction を落とします。

http_requests データセットの入力レコード:

{
  "ClientIP": "203.0.113.42",
  "ClientRequestHost": "example.com",
  "ClientRequestMethod": "POST",
  "ClientRequestPath": "/api/checkout",
  "ClientRequestUserAgent": "curl/7.85.0",
  "EdgeResponseStatus": 502,
  "EdgeStartTimestamp": "2026-05-21T15:00:00Z",
  "RayID": "8e2a1c60ef9e1c9a",
  "OriginResponseTime": 3200000000,
  "WAFAction": "unknown"
}

この例では、EdgeStartTimestamp が RFC3339 文字列として配信されることを前提にしています。ジョブがタイムスタンプを Unix ナノ秒で配信する場合は、to_timestamp() のラッパーを外し、代わりに 1e9 で割ってください。

Transformer:

SELECT
  extract(epoch FROM to_timestamp(EdgeStartTimestamp)) AS unix_ts,
  UPPER(ClientRequestMethod) AS method,
  'Cloudflare' AS provider,
  named_struct(
    'host', ClientRequestHost,
    'path', ClientRequestPath,
    'method', ClientRequestMethod
  ) AS request,
  EdgeResponseStatus >= 500 AS is_server_error,
  [ClientRequestHost, ClientRequestPath] AS request_meta
FROM http_requests
WHERE ClientRequestHost NOT IN ('internal.example.com', 'health.example.com')
  AND ClientRequestPath NOT LIKE '/healthz%'
  AND ClientRequestPath NOT LIKE '/metrics%'

配信されるレコード:

{
  "is_server_error": true,
  "method": "POST",
  "provider": "Cloudflare",
  "request": {
    "host": "example.com",
    "method": "POST",
    "path": "/api/checkout"
  },
  "request_meta": ["example.com", "/api/checkout"],
  "unix_ts": 1779375600
}

バッチ処理中に transformer が失敗すると、そのバッチは失敗します。そのバッチは何も配信されず、Logpush ジョブにエラーが記録され、Logpush は通常のスケジュールでバッチを再試行します。未加工ログへの自動フォールバックはありません。

失敗が続くと、影響を受けたバッチのレコードは最終的に破棄され、復旧できません。

最後のエラーは、ジョブの last_error フィールドに表示されます。よくある原因:

• 出力がサイズ上限を超えた。 レコードあたりの出力を減らすか、WHERE でより多くのレコードを除外してください。
• Cloudflare 内部エラーが発生した。 ジョブ ID とタイムスタンプを添えて Cloudflare Support に連絡してください。

デバッグするには、Transformer Studio で transformer を開き、Run ボタンでサンプルレコードに対してプレビューします。検証と実行のロジックは同じなので、本番で見える問題は、多くの場合プレビューでも再現します。

• Logpush データセット - SQL クエリが参照するフィールド
• Logpush ジョブのセットアップ - Logpush ジョブの作成と管理
• ログフィールドリファレンス - データセット横断のフィールド説明
• フィルター - SQL を使わない簡易なフィルタリング