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 returnNULLto 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
_AllLogsview, 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', )])]>
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', )])]>
Ingestion Results
|
event_id |
user_id |
event_name |
status |
|---|---|---|---|
|
|
|
|
|
|
|
|
|
|
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
- =ERROR’), (‘language’, ”), (‘caption’, )])]>
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._AllLogs`rnWHERE log_name LIKE '%managed_routine_output%'rnORDER BY timestamp DESCrnLIMIT 10;"), ('language', ''), ('caption', )])]>
Log Analytics output
|
timestamp |
severity |
error_type |
missing_key |
raw_payload |
|---|---|---|---|---|
|
|
|
|
|
|
|
|
|
|
null |
|
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._AllLogs`rnWHERE 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_events`rnORDER BY event_id;"), ('language', ''), ('caption', )])]>
Final clean_events Table Result:
|
event_id |
user_id |
event_name |
status |
|
|
|
|
|
|
|
|
|
|
|
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.