Datadog is built for real-time diagnosis. BigQuery, meanwhile, is built for large-scale analytical querying, long retention, and joining telemetry against data that lives outside your observability platform. Moving a filtered subset of APM logs from Datadog to BigQuery is a reasonable thing to want. The path itself is short: a forwarder, a buffer, and a managed sink.
This guide covers that path, with the failure modes that are easy to hit and hard to diagnose called out where they occur.
Architecture for Moving Datadog to BigQuery
Logs are collected by Datadog agents and processed by an ingest-time pipeline. A log forwarding destination then pushes matching logs to an HTTP endpoint. From there, a Cloud Run function receives the payload, projects each log onto the target schema, and publishes to Pub/Sub. A Pub/Sub subscription with a BigQuery delivery type streams those messages into the table, and a dead-letter topic catches anything BigQuery rejects.
The design decision worth explaining is where schema conformance happens. It belongs in the forwarder, not in Datadog. Datadog’s processors are good at parsing and enrichment but awkward at producing an exact field set, and you cannot unit-test a Datadog pipeline. Doing the projection in Python means the mapping is version-controlled, testable, and fixable without touching the observability config.
Step 1: Capture a Real Payload
First, before building anything, get an actual log off the wire.
Confirm in APM > Traces that the services you care about are producing spans, then filter Log Explorer to the logs you intend to export, for example source:apm @dd.trace_id:*.
One caution about the JSON tab in Log Explorer: what it renders is the Logs API representation, which wraps fields in a content object. That envelope is not what pipeline processors operate on, and it is not necessarily what arrives at your HTTP destination either. Treat it as a guide to which fields exist, not as the shape of the payload.
Instead, the authoritative sample is the one your endpoint actually receives. Deploy the function from Step 3 with a temporary handler that logs the raw body, send a small volume through, and build the mapping against that.
Step 2: Parse and Enrich in Datadog
Create a pipeline under Logs > Configuration > Pipelines with a filter matching only the logs you intend to export. This is also the point where most Datadog to BigQuery mapping problems get introduced, so test the filter in Log Explorer before saving.
Grok Parsing
Datadog’s Grok implementation uses the syntax %{MATCHER:EXTRACT:FILTER} with its own matcher names. These are not the Logstash or Elastic names. Datadog uses lowercase identifiers such as notSpace, word, integer, number, ipv4, data, date("pattern") and regex("pattern"). Patterns written with IPORHOST, NOTSPACE, HTTPDATE, WORD or INT will not resolve.
For a standard combined access log in the message field:
access_log %{ipv4:network.client.ip} %{notSpace:http.ident} %{notSpace:http.auth} \[%{date("dd/MMM/yyyy:HH:mm:ss Z"):http.request_time}\] "%{word:http.method} %{notSpace:http.url_details.path}(?: HTTP/%{number:http.version})?" %{integer:http.status_code}
Two things to note. The matcher integer produces an actual integer rather than a string. That matters because a string arriving at an INTEGER column in BigQuery is rejected, and the row goes to the dead-letter topic. That failure presents as a delivery problem rather than a parsing problem, which is why it costs people an afternoon.
The date matcher outputs epoch milliseconds. If you feed that into a TIMESTAMP column expecting RFC 3339, it will either fail or land at some point in 1970. Step 3 handles the conversion explicitly.
Attribute names here follow Datadog’s standard naming, so that its own facets and dashboards work. The forwarder, meanwhile, flattens them to the BigQuery column names.
Test With the Sample
Use the pipeline editor’s preview with the payload from Step 1 and confirm each field extracts. Adjust for your log variations before moving on.
A Note on Field Whitelisting
Some guides suggest a processor that keeps only a named set of attributes, as a way of making the outgoing JSON match the BigQuery schema exactly. I could not find such a processor in Datadog’s processor list. If your account has one, it is a reasonable belt-and-braces addition. If not, do not go looking: the projection in Step 3 does this job better, because it is explicit and testable.
Step 3: The Forwarder
This function terminates the HTTP request, authenticates it, projects each log onto the target schema, and publishes to Pub/Sub.
Authentication and a Contradiction to Avoid
Cloud Run functions offer a “require authentication” setting. It requires the caller to present a Google-signed identity token. Datadog’s log forwarding can send arbitrary custom headers but cannot mint Google identity tokens, so a function with that setting enabled will reject every Datadog request before your code runs.
Two workable options:
- Allow unauthenticated invocations and rely on a strong pre-shared secret in a custom header, compared in constant time. Simplest, and adequate if the secret is long, stored in Secret Manager and rotated.
- Put the function behind an API Gateway or an external HTTPS load balancer with Cloud Armour, and let that layer handle authentication and rate limiting. More moving parts, better posture, and it gives you somewhere to enforce IP allowlisting against Datadog’s published egress ranges.
Pick one deliberately. Otherwise, the failure mode of picking neither is silent 403s at the Google edge that never reach your logs.
main.py
python
import hmac
import json
import os
import urllib.request
from datetime import datetime, timezone
import functions_framework
from google.cloud import pubsub_v1
_METADATA_URL = "http://metadata.google.internal/computeMetadata/v1/project/project-id"
# Columns in the BigQuery table. Anything not listed is dropped.
_SCALAR_FIELDS = (
"id", "host", "service", "message",
"client_ip", "request_method", "request_path",
)
# Tags promoted from the Datadog tag array into their own columns.
_PROMOTED_TAGS = (
"env", "image_name", "image_tag", "container_id", "container_name",
"kube_namespace", "kube_deployment", "kube_cluster_name", "kube_node",
"version",
)
def _resolve_project_id():
"""GCP_PROJECT is a legacy 1st gen variable and is not populated in
2nd gen runtimes. Resolve explicitly, then fall back to metadata."""
for name in ("PUBSUB_PROJECT_ID", "GOOGLE_CLOUD_PROJECT"):
value = os.environ.get(name)
if value:
return value
request = urllib.request.Request(
_METADATA_URL, headers={"Metadata-Flavor": "Google"}
)
with urllib.request.urlopen(request, timeout=2) as response:
return response.read().decode("utf-8")
PROJECT_ID = _resolve_project_id()
TOPIC_ID = os.environ["PUBSUB_TOPIC_ID"]
SHARED_SECRET = os.environ["DATADOG_SECRET"]
_batch_settings = pubsub_v1.types.BatchSettings(
max_messages=100, max_bytes=1024 * 1024, max_latency=0.05
)
publisher = pubsub_v1.PublisherClient(batch_settings=_batch_settings)
TOPIC_PATH = publisher.topic_path(PROJECT_ID, TOPIC_ID)
def _to_rfc3339(value):
"""Datadog may emit ISO 8601 or epoch milliseconds depending on the
field and the processors applied. Normalise both to RFC 3339."""
if value is None:
return None
if isinstance(value, (int, float)):
seconds = value / 1000.0 if value > 1e11 else float(value)
return datetime.fromtimestamp(seconds, tz=timezone.utc).isoformat()
return str(value)
def _parse_tags(raw_tags):
"""Datadog tags arrive as a list of 'key:value' strings, or occasionally
as a comma-separated string. Return the list plus a promoted dict."""
if isinstance(raw_tags, str):
raw_tags = [t for t in raw_tags.split(",") if t]
elif not isinstance(raw_tags, list):
raw_tags = []
promoted = {}
for tag in raw_tags:
if ":" not in tag:
continue
key, _, value = tag.partition(":")
if key in _PROMOTED_TAGS and key not in promoted:
promoted[key] = value
return raw_tags, promoted
def _coerce_int(value):
try:
return int(value)
except (TypeError, ValueError):
return None
def project(log):
"""Map one Datadog log onto the BigQuery schema. Every column the table
declares is produced here, and nothing else is."""
attributes = log.get("attributes") or {}
tags, promoted = _parse_tags(log.get("tags"))
row = {name: log.get(name) or attributes.get(name) for name in _SCALAR_FIELDS}
row["timestamp"] = _to_rfc3339(log.get("timestamp") or attributes.get("timestamp"))
row["status_code"] = _coerce_int(log.get("status_code") or attributes.get("status_code"))
row["tags"] = tags
row["attributes"] = json.dumps(attributes) if attributes else None
for name in _PROMOTED_TAGS:
row[name] = promoted.get(name)
return row
def _is_publishable(row):
"""timestamp and service are REQUIRED in BigQuery. Rows missing them
would be dead-lettered, so drop them here where it is visible."""
return bool(row.get("timestamp")) and bool(row.get("service"))
@functions_framework.http
def handle_log_request(request):
if request.method != "POST":
return "Only POST is accepted", 405
received = request.headers.get("X-Datadog-Secret", "")
if not hmac.compare_digest(received, SHARED_SECRET):
return "Unauthorised", 403
payload = request.get_json(silent=True)
if payload is None:
return "Invalid or empty JSON body", 400
entries = payload if isinstance(payload, list) else [payload]
futures, dropped = [], 0
for entry in entries:
if not isinstance(entry, dict):
dropped += 1
continue
row = project(entry)
if not _is_publishable(row):
dropped += 1
print(json.dumps({"event": "dropped_incomplete", "id": row.get("id")}))
continue
data = json.dumps(row, separators=(",", ":")).encode("utf-8")
if len(data) > 9_000_000: # Pub/Sub message limit is 10MB
dropped += 1
print(json.dumps({"event": "dropped_oversized", "id": row.get("id")}))
continue
attrs = {"dd_log_id": str(row["id"])} if row.get("id") else {}
futures.append(publisher.publish(TOPIC_PATH, data=data, **attrs))
# Resolve after the loop. Calling result() inside it serialises every
# publish and turns a large batch into a timeout.
published, failed = 0, 0
for future in futures:
try:
future.result(timeout=30)
published += 1
except Exception as exc:
failed += 1
print(json.dumps({"event": "publish_failed", "error": type(exc).__name__}))
print(json.dumps({
"event": "batch_complete", "received": len(entries),
"published": published, "dropped": dropped, "failed": failed,
}))
if failed:
return f"published={published} failed={failed}", 500
return f"published={published} dropped={dropped}", 200requirements.txt
functions-framework
google-cloud-pubsub
Deployment Notes
Runtime: Python 3.11 or newer, entry point handle_log_request, 2nd gen. Give it a dedicated service account with roles/pubsub.publisher scoped to the topic and nothing else.
Environment variables: PUBSUB_TOPIC_ID set to your topic ID, and DATADOG_SECRET mounted from Secret Manager rather than pasted as a plain environment variable. Do not attempt to set GCP_PROJECT, which is a reserved name.
Both TOPIC_ID and DATADOG_SECRET are read with os.environ[...] rather than .get(), so a misconfigured deployment fails at startup instead of at 3 am.
On Duplicates
Returning 500 causes Datadog to retry, and anything already published in that batch gets published again. There is no deduplication on the streaming path into BigQuery. Two options exist here. The simpler choice is to accept it and deduplicate at read time using the Datadog log ID. A more involved alternative is tracking published IDs in Memorystore and skipping repeats, which is more machinery than most pipelines need.
The read-time version:
CREATE OR REPLACE VIEW `PROJECT.apm_logs.v_datadog_apm_traces` AS
SELECT * EXCEPT(row_num) FROM (
SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY timestamp) AS row_num
FROM `PROJECT.apm_logs.datadog_apm_traces`
)
WHERE row_num = 1;In short, query the view, not the table.
Step 4: Pub/Sub and BigQuery Setup
Create a topic, for example datadog-apm-logs-topic, with default settings. This is the buffer that decouples Datadog to BigQuery delivery from BigQuery’s own availability.
Then the dataset and table. Schema:
[
{ "name": "id", "type": "STRING", "mode": "NULLABLE" },
{ "name": "timestamp", "type": "TIMESTAMP", "mode": "REQUIRED" },
{ "name": "host", "type": "STRING", "mode": "NULLABLE" },
{ "name": "service", "type": "STRING", "mode": "REQUIRED" },
{ "name": "message", "type": "STRING", "mode": "NULLABLE" },
{ "name": "tags", "type": "STRING", "mode": "REPEATED" },
{ "name": "attributes", "type": "STRING", "mode": "NULLABLE" },
{ "name": "client_ip", "type": "STRING", "mode": "NULLABLE" },
{ "name": "request_method", "type": "STRING", "mode": "NULLABLE" },
{ "name": "request_path", "type": "STRING", "mode": "NULLABLE" },
{ "name": "status_code", "type": "INTEGER", "mode": "NULLABLE" },
{ "name": "env", "type": "STRING", "mode": "NULLABLE" },
{ "name": "image_name", "type": "STRING", "mode": "NULLABLE" },
{ "name": "image_tag", "type": "STRING", "mode": "NULLABLE" },
{ "name": "container_id", "type": "STRING", "mode": "NULLABLE" },
{ "name": "container_name", "type": "STRING", "mode": "NULLABLE" },
{ "name": "kube_namespace", "type": "STRING", "mode": "NULLABLE" },
{ "name": "kube_deployment", "type": "STRING", "mode": "NULLABLE" },
{ "name": "kube_cluster_name", "type": "STRING", "mode": "NULLABLE" },
{ "name": "kube_node", "type": "STRING", "mode": "NULLABLE" },
{ "name": "version", "type": "STRING", "mode": "NULLABLE" }
]Three deliberate choices here.attributes is STRING rather than JSON. BigQuery’s JSON type is well supported in queries, but support for JSON columns through the Pub/Sub BigQuery subscription has been inconsistent, and a rejection here dead-letters the whole row. Storing the serialised object and applying PARSE_JSON() in a view costs nothing and removes a class of ingestion failure. If you confirm JSON columns work on your project and Pub/Sub version, switch it.
INTEGER is correct for the JSON schema API. INT64 is the SQL DDL spelling and will be rejected here.
Timestamp and service are REQUIRED, which means any row missing them is rejected and dead-lettered. The forwarder drops such rows before publishing, so the failure is visible in function logs rather than buried in a dead-letter queue.
Partition by timestamp using field partitioning, and cluster on service, status_code and kube_namespace. Set a partition expiration matching your actual retention requirement rather than leaving it unbounded.
Step 5: The BigQuery Subscription
Create a subscription on the topic with delivery type “Write to BigQuery,” pointing at the dataset and table.
Leave “write metadata” unchecked unless you have added the corresponding _MESSAGE_ID, _PUBLISH_TIME and _ATTRIBUTES columns. Enabling “drop unknown fields” is worth considering, since it makes the pipeline tolerant of the forwarder emitting a field the table does not yet have. Without it, schema drift stops ingestion entirely.
Configure a dead-letter topic, for example datadog-apm-logs-dlq, with maximum delivery attempts around 5. Create a subscription on the dead-letter topic too, or the messages will land somewhere nobody is looking. Grant the Pub/Sub service agent publisher rights on the dead-letter topic and subscriber rights on the source subscription, which is a step people routinely miss and which produces silent failure.
On Dataflow: for streaming pre-shaped JSON, the native subscription is simpler and cheaper. Reach for Dataflow when you need windowing, aggregation or stream joins before the data lands.
Step 6: Configure Forwarding in Datadog
Under Logs > Configuration, create a custom HTTP destination. The exact menu path and availability vary by plan, so confirm against your account rather than against any guide.
Set the endpoint to the function URL, method POST, body format JSON, and add a custom header X-Datadog-Secret carrying the same value as the function’s secret. Set the query to the same filter as your pipeline.
One correction to a common misunderstanding: Datadog processing pipelines run at ingest, against their own filter, and apply to every log that matches. They are not selected per destination. Anything matching your pipeline filter is already transformed by the time forwarding sees it, whether or not it is being forwarded.
Step 7: Validate the Datadog to BigQuery Pipeline
Generate traffic, then walk the path in order.
- In Log Explorer, confirm new logs show the parsed attributes.
- In the function’s logs, look for
batch_completeentries and check the dropped and failed counts. Non-zero drops withdropped_incompletemean your projection is not finding a field where it expects it. - In Pub/Sub, confirm the published message count is rising, and the unacknowledged count is falling.
- In BigQuery, query the table.
SELECT timestamp, service, client_ip, request_method, request_path, status_code
FROM `PROJECT.apm_logs.datadog_apm_traces`
WHERE timestamp >= TIMESTAMP(CURRENT_DATE())
ORDER BY timestamp DESC
LIMIT 10;Note the filter. _PARTITIONDATE exists only on ingestion-time partitioned tables, and this table is partitioned on the timestamp field, so that pseudo-column is not available.
Monitoring the Pipeline From Datadog to BigQuery
The single most important alert is on subscription/dead_letter_message_count. A non-zero value means rows are being rejected by BigQuery, and it is almost always a type mismatch or a missing required field. Everything else is secondary: function execution errors and latency, Pub/Sub oldest unacked message age, and BigQuery streaming insert errors.
Alert on the dead-letter count reaching one, not on a threshold. The first rejection tells you something changed upstream, and the second thousand tell you the same thing more expensively.
What Moving Datadog to BigQuery Is Good For, and What It Is Not
The genuine wins are analytical. Correlating performance data against product usage over months, rather than the retention window your observability plan gives you. Historical forensic work across trace data. Reliability reporting over long periods. Joining service behaviour to customer segments or regions in the same query engine as the rest of your business data.
The claim to be careful with is compliance. Exporting logs to a long-retention warehouse does not satisfy privacy obligations; instead, it creates them. Client IP addresses are personal data under GDPR, and this pipeline places them in a store you have described to yourself as cheap to keep forever. That engages the storage limitation principle and the right to erasure. You now need a documented lawful basis, a retention period you actually enforce through partition expiry, and a way to delete a subject’s rows on request. If health data could appear in a log line, HIPAA brings a business associate agreement and a good deal more.
Overall, the right framing is that this improves auditability and analytical reach, and that it adds a data protection surface which needs owning. The cheapest control is upstream: do not export fields you would struggle to justify retaining. Dropping client_ip, or truncating it to a network prefix in the projection function, costs one line and removes most of the problem.
Asked Questions
Design Questions
Why export Datadog APM logs to BigQuery instead of just using Datadog? Datadog is optimised for real-time diagnosis within its own retention window. BigQuery, in contrast, is built for long-term retention and for joining telemetry against data that lives outside your observability platform, such as product usage or customer segments. Moving Datadog to BigQuery lets you do analytical work Datadog itself isn’t designed for.
In a Datadog to BigQuery pipeline, why does schema conformance happen in the Cloud Run forwarder instead of in Datadog? Because Datadog’s processors are good at parsing and enrichment but awkward at producing an exact field set, and Datadog pipelines can’t be unit-tested. Doing the projection in Python makes the field mapping version-controlled, testable, and fixable without touching the observability configuration.
What causes silent failures when exporting Datadog to BigQuery? Two common causes stand out. One is enabling Cloud Run’s “require authentication” setting, which rejects every Datadog request before your code runs, since Datadog can’t mint Google identity tokens. The other is type mismatches, such as a Grok-parsed string landing on an INTEGER column, which sends the row to the dead-letter topic rather than raising a visible parsing error.
Operational Questions
How do I handle duplicate rows in this pipeline? There’s no deduplication on the streaming path into BigQuery, since a 500 response causes Datadog to retry the whole batch. The simplest fix is a read-time view using ROW_NUMBER() partitioned by the Datadog log ID, querying that view instead of the raw table.
Does exporting Datadog logs to BigQuery create compliance obligations? Yes. It doesn’t satisfy privacy obligations; it creates them. Client IP addresses are personal data under GDPR, so a long-retention warehouse copy engages the storage limitation principle and the right to erasure. Dropping or truncating fields like client_ip in the projection step is the cheapest way to reduce that exposure.
Read More Here
