Safety
How to Use Python UDF Logging for Pipeline Observability
We are announcing a new logging feature that is now available in Python User-Defined Function (UDF) in BigQuery. Developers can now use Python’s standard logging library directly inside BigQuery UDFs
We are announcing a new logging feature that is now available in Python User-Defined Function (UDF) in BigQuery. Developers can now use Python’s standard logging library directly inside BigQuery UDFs to emit structured logs seamlessly into Cloud Logging—with zero external plugins or custom connectors required. While UDF logging enables a wide range of operational monitoring and debugging workflows, to demonstrate this feature we are showcasing an example for building fault-tolerant, resilient data pipelines. In the example below, we are capturing row-level diagnostic telemetry during query execution, so that teams can easily isolate problematic records, trigger real-time alerts, and triage issues via SQL without halting entire workloads. In this post, we’ll show you how to get started with Python UDF logging and walk through an e-commerce event ingestion pipeline as a practical, end-to-end example. Architecture overview The workflow consists of three stages: Inline validation: The Python UDF parses incoming JSON payloads. Valid records pass to the destination table; corrupted or non-compliant rows trigger logging.error() and return NULL to prevent query failure. Real-time alerting: A Cloud Monitoring log-based metric notifies your team when error rates breach SLA thresholds. SQL triage & replay: Data engineers query the quarantined payloads in BigQuery using standard SQL over the linked _AllLogs view, diagnose the root cause, and re-ingest recovered data. Although this specific demonstration relies on Python for data validation and transformation, you can seamlessly integrate Python-driven logging and validation into pipelines that perform SQL-based transformations. Quickstart walkthrough: Step-by-step Step 1: Create the generic Python UDF This UDF parses JSON and checks for a required field. If parsing fails or the required key is missing, it logs structured diagnostic JSON and returns NULL. code_block <ListValue: [StructValue([('code', 'CREATE OR REPLACE FUNCTION YOUR_PROJECT.YOUR_DATASET.safe_parse_json(rn raw_payload STRING,rn required_key STRINGrn)rnRETURNS STRINGrnLANGUAGE pythonrnOPTIONS (entry_point=safe_parse_json', runtime_version=python-3.11')rnAS r'''rnimport jsonrnimport loggingrnrndef safe_parse_json(raw_payload, required_key):rn if not raw_payload or raw_payload.strip() == "":rn logging.warning(json.dumps({"error_type": "EMPTY_PAYLOAD"}))rn return Nonernrn try:rn data = json.loads(raw_payload)rn if required_key and (required_key not in data or data[required_key] is None):rn logging.error(json.dumps({rn "error_type": "MISSING_REQUIRED_KEY",rn "missing_key": required_key,rn "raw_payload": raw_payload[:200]rn }))rn return Nonern return json.dumps(data)rnrn except Exception as e:rn logging.error(json.dumps({rn "error_type": "JSON_SYNTAX_ERROR",rn "error_message": str(e),rn "raw_payload": raw_payload[:200]rn }))rn return Nonern''';'), ('language', ''), ('caption', <wagtail.rich_text.RichText object at 0x7fbf92f2ab10>)])]> Step 2: Run ingestion with logging enabled In this example, we simulate an e-commerce event ingestion pipeline processing incoming user actions (click, view, checkout). Our goal is to extract valid events into our clean analytics table while gracefully quarantining broken or missing-schema payloads. Enable logging in your session with SET @@udf.properties: code_block <ListValue: [StructValue([('code', 'SET @@udf.properties = JSON '{"enableDebugOutput": true}';rnrnCREATE OR REPLACE TABLE YOUR_PROJECT.YOUR_DATASET.clean_events ASrnWITH sample_events AS (rn SELECT evt_001' AS event_id, '{"user_id": "usr_101", "event": "click", "amount": 25.5}' AS raw_data UNION ALLrn SELECT evt_002' AS event_id, corrupted_json{' AS raw_data UNION ALL -- Syntax errorrn SELECT evt_003' AS event_id, '{"event": "view"}' AS raw_data UNION ALL -- Missing required user_idrn SELECT evt_004' AS event_id, '{"user_id": "usr_102", "event": "checkout", "amount": 99.0}' AS raw_datarn),rnparsed_events AS (rn SELECTrn event_id,rn YOUR_PROJECT.YOUR_DATASET.safe_parse_json(raw_data, user_id') AS validated_jsonrn FROM sample_eventsrn)rnSELECTrn event_id,rn JSON_VALUE(validated_json, '.user_id') AS user_id,rn JSON_VALUE(validated_json, '.event') AS event_name,rn VALID' AS statusrnFROM parsed_eventsrnWHERE validated_json IS NOT NULL;'), ('language', ''), ('caption', <wagtail.rich_text.RichText object at 0x7fbf92f29d90>)])]> Ingestion Results event_id user_id event_name status evt_001 usr_101 click VALID evt_004 usr_102 checkout VALID The query succeeds without interruption while bad rows (2 and 3) are routed to Cloud Logging. Step 3: Set up real-time alerts on Cloud Logging You can configure alerts to notify your on-call team as soon as malformed payloads exceed acceptable thresholds. In the Google Cloud Console (1-Click) In the Google Cloud Console, open Logs Explorer. Paste the log filter for your UDF routine: code_block <ListValue: [StructValue([('code', 'resource.type="bigquery.googleapis.com/ManagedRoutineInvocation"rnseverity>=ERROR'), ('language', ''), ('caption', <wagtail.rich_text.RichText object at 0x7fbf92606810>)])]> 3. Click Create alert in the top action bar. 4. Set your notification channels (e.g., Slack, PagerDuty, or Email) and save the policy. Step 4: Triage bad rows with BigQuery Log Analytics "Note: To query your logs via BigQuery SQL, ensure you have enabled BigQuery Log Analytics on your Cloud Logging bucket and created a linked dataset." Query the quarantined payloads using BigQuery SQL via the linked _AllLogs view: code_block <ListValue: [StructValue([('code', "SELECTrn timestamp,rn severity,rn JSON_VALUE(json_payload.error_type) AS error_type,rn JSON_VALUE(json_payload.missing_key) AS missing_key,rn JSON_VALUE(json_payload.raw_payload) AS raw_payloadrnFROM YOUR_PROJECT.YOUR_LOGS_LINKED_DATASET._AllLogsrnWHERE log_name LIKE '%managed_routine_output%'rnORDER BY timestamp DESCrnLIMIT 10;"), ('language', ''), ('caption', <wagtail.rich_text.RichText object at 0x7fbf939bf9d0>)])]> Log Analytics output timestamp severity error_type missing_key raw_payload 2026-09-17 18:35:50 ERROR MISSING_REQUIRED_KEY user_id {"event": "view"} 2026-09-17 18:35:50 ERROR JSON_SYNTAX_ERROR null corrupted_json{ Recovering data Because full raw payloads are preserved in Cloud Logging, you can recover quarantined records directly with SQL once upstream data is fixed: code_block <ListValue: [StructValue([('code', "-- 1. Replay and recover quarantined rows from Cloud Logging sink tablernINSERT INTO YOUR_PROJECT.YOUR_DATASET.clean_events (event_id, user_id, event_name, status)rnSELECTrn 'evt_003_recovered' AS event_id,rn 'default_user' AS user_id, -- Apply fallback default for the missing fieldrn JSON_VALUE(JSON_VALUE(json_payload.raw_payload), '$.event') AS event_name,rn 'RECOVERED' AS statusrnFROM YOUR_PROJECT.YOUR_LOGS_LINKED_DATASET._AllLogsrnWHERE JSON_VALUE(json_payload.error_type) = 'MISSING_REQUIRED_KEY'rnrn-- 2. Inspect the final clean destination tablernSELECT event_id, user_id, event_name, statusrnFROM YOUR_PROJECT.YOUR_DATASET.clean_eventsrnORDER BY event_id;"), ('language', ''), ('caption', <wagtail.rich_text.RichText object at 0x7fbf939bd950>)])]> Final clean_events Table Result: event_id user_id event_name status evt_001 usr_101 click VALID evt_004 usr_102 checkout VALID evt_003_recovered default_user view RECOVERED All valid and recovered data is now unified in the production table without any data loss. Other use cases Shadow execution & safe migrations: Run legacy and new business logic side-by-side in a UDF. Return the legacy result to production while logging calculation differences to Cloud Logging. PII redaction auditing: Mask sensitive tokens (such as SSNs or account numbers) before writing to BigQuery, while emitting anonymized audit events to Cloud Logging. ML inference drift: Log warnings when embedded scoring logic encounters out-of-vocabulary inputs or low model confidence. Best practices Enable debug output: Include SET @@udf.properties = JSON '{"enableDebugOutput": true}'; in the session where you want logs captured. Avoid sensitive data in logs: Cloud Logging is visible to users with Logging Viewer permissions. Do not log credentials or secrets. Cost management: On high-throughput pipelines, log selectively (e.g., only on errors or via sampling) to manage logging ingestion volume. Getting started BigQuery Python UDF Documentation BigQuery Log Analytics Overview
Related
Source: Google Cloud AI | 2026-10-19