Great question — lots of teams do this by syncing Prefect’s core run tables to BigQuery and building a few reusable views. Since you’re on Prefect 3.x, you can rely on the denormalized state fields on runs (

state_type

,

state_name

,

start_time

,

end_time

,

expected_start_time

, etc.) for fast analytics without always joining state history. Below is a starter pack of BigQuery SQL for: - Success/error rates over time - Throughput (runs started/completed) - Queue latency (scheduled → start) - Run duration (start → end) with percentiles - Failure diagnostics (top error messages) - Pool/queue breakdown - Task-level failure rate Assumptions and tips - Replace

YOUR_PROJECT.YOUR_DATASET

with your dataset. Table names are commonly

flow_run

,

flow_run_state

,

task_run

,

task_run_state

,

deployment

. If your sync pluralized tables, update names accordingly. - If you only want top-level flows (exclude subflows), add

parent_task_run_id IS NULL

to filters. - Terminal states:

COMPLETED, FAILED, CANCELLED, CRASHED

. Quick schema check (optional)

-- List tables
SELECT table_name
FROM `YOUR_PROJECT.YOUR_DATASET`.INFORMATION_SCHEMA.TABLES;

-- Inspect a table
SELECT column_name, data_type
FROM `YOUR_PROJECT.YOUR_DATASET`.INFORMATION_SCHEMA.COLUMNS
WHERE table_name = 'flow_run';

1) Daily success / error rates (top-level flows)

WITH base AS (
  SELECT
    DATE(COALESCE(end_time, start_time, created)) AS day,
    state_type
  FROM `YOUR_PROJECT.YOUR_DATASET.flow_run`
  WHERE
    parent_task_run_id IS NULL
    AND end_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 90 DAY)  -- lookback window
    -- consider only terminal runs for rate denominators
    AND state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED')
)
SELECT
  day,
  COUNT(*) AS terminal_runs,
  COUNTIF(state_type = 'COMPLETED') AS completed,
  COUNTIF(state_type IN ('FAILED','CRASHED')) AS errors,
  SAFE_DIVIDE(COUNTIF(state_type = 'COMPLETED'), COUNT(*)) AS success_rate,
  SAFE_DIVIDE(COUNTIF(state_type IN ('FAILED','CRASHED')), COUNT(*)) AS error_rate
FROM base
GROUP BY day
ORDER BY day;

2) Throughput (runs started vs completed per day)

SELECT
  d AS day,
  -- started
  (SELECT COUNT(*) FROM `YOUR_PROJECT.YOUR_DATASET.flow_run`
   WHERE parent_task_run_id IS NULL
     AND DATE(start_time) = d) AS runs_started,
  -- completed (terminal)
  (SELECT COUNT(*) FROM `YOUR_PROJECT.YOUR_DATASET.flow_run`
   WHERE parent_task_run_id IS NULL
     AND state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED')
     AND DATE(end_time) = d) AS runs_completed
FROM UNNEST(GENERATE_DATE_ARRAY(
       DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY), CURRENT_DATE())) AS d
ORDER BY day;

3) Queue latency and run duration percentiles (daily) - Queue latency:

start_time - expected_start_time

- Run duration:

end_time - start_time

``` WITH base AS ( SELECT DATE(COALESCE(end_time, start_time, created)) AS day, TIMESTAMP_DIFF(start_time, expected_start_time, SECOND) AS queue_latency_s, TIMESTAMP_DIFF(end_time, start_time, SECOND) AS run_duration_s FROM

YOUR_PROJECT.YOUR_DATASET.flow_run

WHERE parent_task_run_id IS NULL AND end_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY) AND state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED') ) SELECT day, -- Queue latency percentiles (only where both timestamps exist) (SELECT APPROX_QUANTILES(x, 100)[OFFSET(50)] FROM UNNEST(ARRAY(SELECT queue_latency_s FROM base b WHERE b.day = base.day AND queue_latency_s IS NOT NULL)) x) AS ql_p50_s, (SELECT APPROX_QUANTILES(x, 100)[OFFSET(95)] FROM UNNEST(ARRAY(SELECT queue_latency_s FROM base b WHERE b.day = base.day AND queue_latency_s IS NOT NULL)) x) AS ql_p95_s, -- Run duration percentiles (SELECT APPROX_QUANTILES(x, 100)[OFFSET(50)] FROM UNNEST(ARRAY(SELECT run_duration_s FROM base b WHERE b.day = base.day AND run_duration_s IS NOT NULL)) x) AS dur_p50_s,

(SELECT APPROX_QUANTILES(x, 100)[OFFSET(95)] FROM UNNEST(ARRAY(SELECT run_duration_s FROM base b WHERE b.day = base.day AND run_duration_s IS NOT NULL)) x) AS dur_p95_s FROM base GROUP BY day ORDER BY day;

4) Current backlog (scheduled but not started)

SELECT work_pool_name, work_queue_name, COUNT(*) AS scheduled_not_started FROM

YOUR_PROJECT.YOUR_DATASET.flow_run

WHERE state_type = 'SCHEDULED' AND (start_time IS NULL OR start_time > CURRENT_TIMESTAMP()) GROUP BY work_pool_name, work_queue_name ORDER BY scheduled_not_started DESC;

5) Failure diagnostics: top error messages (from latest state record)

WITH latest_state AS ( SELECT flow_run_id, name AS state_name, message, type AS state_type, timestamp, ROW_NUMBER() OVER (PARTITION BY flow_run_id ORDER BY timestamp DESC) AS rn FROM

YOUR_PROJECT.YOURDATASET.flow_run_state

) SELECT fr.flow_id, fr.deployment_id, fr.name AS flow_run_name, ls.message, COUNT(*) AS occurrences FROM

YOUR_PROJECT.YOUR_DATASET.flow_run

fr JOIN latest_state ls ON ls.flow_run_id = fr.id AND ls.rn = 1 WHERE fr.parent_task_run_id IS NULL AND ls.type IN ('FAILED','CRASHED') AND fr.end_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY) GROUP BY flow_id, deployment_id, flow_run_name, message ORDER BY occurrences DESC LIMIT 100;

6) Pool/queue performance (success rate and duration by work pool/queue)

WITH base AS ( SELECT work_pool_name, work_queue_name, state_type, TIMESTAMP_DIFF(end_time, start_time, SECOND) AS run_duration_s FROM

YOUR_PROJECT.YOUR_DATASET.flow_run

WHERE parent_task_run_id IS NULL AND end_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY) AND state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED') ) SELECT work_pool_name, work_queue_name, COUNT(*) AS terminal_runs, COUNTIF(state_type = 'COMPLETED') AS completed, SAFE_DIVIDE(COUNTIF(state_type = 'COMPLETED'), COUNT(*)) AS success_rate, AVG(run_duration_s) AS avg_run_duration_s, APPROX_QUANTILES(run_duration_s, 100)[OFFSET(95)] AS p95_run_duration_s FROM base GROUP BY work_pool_name, work_queue_name ORDER BY terminal_runs DESC;

7) Flow/deployment leaderboard (volume and error rate)

SELECT fr.flow_id, fr.deployment_id, COUNT(*) AS runs, COUNTIF(fr.state_type IN ('FAILED','CRASHED')) AS failed_runs, SAFE_DIVIDE(COUNTIF(fr.state_type IN ('FAILED','CRASHED')), COUNT(*)) AS error_rate FROM

YOUR_PROJECT.YOUR_DATASET.flow_run

fr WHERE fr.parent_task_run_id IS NULL AND fr.end_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY) AND fr.state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED') GROUP BY flow_id, deployment_id ORDER BY runs DESC;

8) Task-run failure rate (optional, deeper diagnostics)

SELECT DATE(COALESCE(end_time, start_time, created)) AS day, COUNT(*) AS task_runs, COUNTIF(state_type IN ('FAILED','CRASHED')) AS failed_task_runs, SAFE_DIVIDE(COUNTIF(state_type IN ('FAILED','CRASHED')), COUNT(*)) AS task_error_rate FROM

YOUR_PROJECT.YOUR_DATASET.task_run

WHERE end_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY) GROUP BY day ORDER BY day;

Recommended views to create for reuse
- A base view that adds convenience fields:

CREATE OR REPLACE VIEW

YOUR_PROJECT.YOUR_DATASET.view_flow_run_metrics

AS SELECT id, flow_id, deployment_id, work_pool_name, work_queue_name, parent_task_run_id, created, start_time, end_time, expected_start_time, state_type, state_name, TIMESTAMP_DIFF(start_time, expected_start_time, SECOND) AS queue_latency_s, TIMESTAMP_DIFF(end_time, start_time, SECOND) AS run_duration_s, DATE(COALESCE(end_time, start_time, created)) AS run_day FROM `YOUR_PROJECT.YOUR_DATASET.flow_run`; ``` - Then your time series become simpler by grouping/selecting from

view_flow_run_metrics

. Looker/Looker Studio starter - Connect BigQuery → your dataset.

- Use

view_flow_run_metrics

as the primary source. - Dimensions:

run_day

,

work_pool_name

,

work_queue_name

,

deployment_id

,

flow_id

. - Metrics: - Success rate:

COUNTIF(state_type = 'COMPLETED') / COUNTIF(state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED'))

- Error rate:

COUNTIF(state_type IN ('FAILED','CRASHED')) / COUNTIF(terminal)

- Throughput:

COUNT(*)

grouped by

run_day

(use filters for started vs completed if you model both) - Queue latency (avg, P95): use

queue_latency_s

- Run duration (avg, P95): use

run_duration_s

- Suggested charts: - Time series: success rate by day - Time series: runs completed by day (throughput) - Time series: P95 run duration and queue latency - Tables: Top flows/deployments by volume, by error rate - Table: Pool/queue performance Questions to tailor this for you - What are your exact table names in BigQuery (singular vs plural)? I can adjust the queries. - Do you want to include subflows or only top-level runs? - Any specific groupings (per flow, per deployment, per work pool/queue) you want prioritized? - Time window defaults (7/14/30/90 days)? If you share your dataset and preferred breakdowns, I’m happy to turn these into ready-to-run views and a Looker Studio template.

Awesome — thanks for the details. Below is a tailored starter kit for BigQuery with singular table names, only top-level runs, grouped by flow, and filterable by deployment name. I’ll use a master view to simplify everything, then a set of queries with a days_lookback parameter (set to 7/14/30/90 as you prefer). Replace PROJECT.DATASET below with your values. 1) Create a reusable metrics view (top-level runs only, includes flow and deployment names)

CREATE OR REPLACE VIEW `PROJECT.DATASET.view_flow_run_metrics_top_level` AS
SELECT
  fr.id,
  fr.flow_id,
  f.name AS flow_name,
  fr.deployment_id,
  d.name AS deployment_name,
  fr.name AS flow_run_name,
  fr.created,
  fr.start_time,
  fr.end_time,
  fr.expected_start_time,
  fr.state_type,
  fr.state_name,
  -- convenience metrics
  IF(fr.start_time IS NOT NULL AND fr.expected_start_time IS NOT NULL,
     TIMESTAMP_DIFF(fr.start_time, fr.expected_start_time, SECOND), NULL) AS queue_latency_s,
  IF(fr.end_time IS NOT NULL AND fr.start_time IS NOT NULL,
     TIMESTAMP_DIFF(fr.end_time, fr.start_time, SECOND), NULL) AS run_duration_s,
  DATE(COALESCE(fr.end_time, fr.start_time, fr.created)) AS run_day
FROM `PROJECT.DATASET.flow_run` fr
LEFT JOIN `PROJECT.DATASET.flow` f ON f.id = fr.flow_id
LEFT JOIN `PROJECT.DATASET.deployment` d ON d.id = fr.deployment_id
WHERE fr.parent_task_run_id IS NULL;

2) Daily success/error rates by flow (filterable by deployment name) - Set days_lookback to 7, 14, 30, or 90

DECLARE days_lookback INT64 DEFAULT 30;
DECLARE deployment_filter STRING DEFAULT NULL;  -- set to a name to filter, e.g. 'my-deployment'

WITH base AS (
  SELECT *
  FROM `PROJECT.DATASET.view_flow_run_metrics_top_level`
  WHERE end_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL days_lookback DAY)
    AND state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED')
    AND (deployment_filter IS NULL OR deployment_name = deployment_filter)
)
SELECT
  run_day,
  flow_name,
  COUNT(*) AS terminal_runs,
  COUNTIF(state_type = 'COMPLETED') AS completed,
  COUNTIF(state_type IN ('FAILED','CRASHED')) AS errors,
  SAFE_DIVIDE(COUNTIF(state_type = 'COMPLETED'), COUNT(*)) AS success_rate,
  SAFE_DIVIDE(COUNTIF(state_type IN ('FAILED','CRASHED')), COUNT(*)) AS error_rate
FROM base
GROUP BY run_day, flow_name
ORDER BY run_day, flow_name;

3) Throughput (started vs completed per day) by flow

DECLARE days_lookback INT64 DEFAULT 30;
DECLARE deployment_filter STRING DEFAULT NULL;

WITH base AS (
  SELECT *
  FROM `PROJECT.DATASET.view_flow_run_metrics_top_level`
  WHERE (start_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL days_lookback DAY)
         OR end_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL days_lookback DAY))
    AND (deployment_filter IS NULL OR deployment_name = deployment_filter)
),
calendar AS (
  SELECT d AS run_day
  FROM UNNEST(GENERATE_DATE_ARRAY(
    DATE_SUB(CURRENT_DATE(), INTERVAL days_lookback DAY), CURRENT_DATE()
  )) AS d
)
SELECT
  c.run_day,
  b.flow_name,
  -- runs started on this day
  COUNTIF(DATE(b.start_time) = c.run_day) AS runs_started,
  -- runs that reached a terminal state on this day
  COUNTIF(DATE(b.end_time) = c.run_day AND b.state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED')) AS runs_completed
FROM calendar c
LEFT JOIN base b
  ON (DATE(b.start_time) = c.run_day OR DATE(b.end_time) = c.run_day)
GROUP BY c.run_day, b.flow_name
ORDER BY c.run_day, b.flow_name;

4) Latency and duration percentiles per day by flow ``` DECLARE days_lookback INT64 DEFAULT 30; DECLARE deployment_filter STRING DEFAULT NULL; WITH base AS ( SELECT * FROM

PROJECT.DATASET.view_flow_run_metrics_top_level

WHERE end_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL days_lookback DAY) AND state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED') AND (deployment_filter IS NULL OR deployment_name = deployment_filter) ), per_day_flow AS ( SELECT run_day, flow_name,

ARRAY_AGG(queue_latency_s IGNORE NULLS) AS ql_arr, ARRAY_AGG(run_duration_s IGNORE NULLS) AS dur_arr FROM base GROUP BY run_day, flow_name ) SELECT run_day, flow_name, -- queue latency percentiles IF(ARRAY_LENGTH(ql_arr) > 0, APPROX_QUANTILES(ql_arr, 100)[OFFSET(50)], NULL) AS ql_p50_s, IF(ARRAY_LENGTH(ql_arr) > 0, APPROX_QUANTILES(ql_arr, 100)[OFFSET(95)], NULL) AS ql_p95_s, -- run duration percentiles IF(ARRAY_LENGTH(dur_arr) > 0, APPROX_QUANTILES(dur_arr, 100)[OFFSET(50)], NULL) AS dur_p50_s, IF(ARRAY_LENGTH(dur_arr) > 0, APPROX_QUANTILES(dur_arr, 100)[OFFSET(95)], NULL) AS dur_p95_s FROM per_day_flow ORDER BY run_day, flow_name;

5) Flow leaderboard (volume and error rate), last N days, filterable by deployment name

DECLARE days_lookback INT64 DEFAULT 30; DECLARE deployment_filter STRING DEFAULT NULL; WITH base AS ( SELECT * FROM

PROJECT.DATASET.view_flow_run_metrics_top_level

WHERE end_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL days_lookback DAY) AND state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED') AND (deployment_filter IS NULL OR deployment_name = deployment_filter) ) SELECT flow_name, COUNT(*) AS runs, COUNTIF(state_type IN ('FAILED','CRASHED')) AS failed_runs, SAFE_DIVIDE(COUNTIF(state_type IN ('FAILED','CRASHED')), COUNT(*)) AS error_rate, AVG(run_duration_s) AS avg_run_duration_s, APPROX_QUANTILES(run_duration_s, 100)[OFFSET(95)] AS p95_run_duration_s FROM base GROUP BY flow_name ORDER BY runs DESC;

6) Failure diagnostics: top error messages by flow (from latest state)

DECLARE days_lookback INT64 DEFAULT 30; DECLARE deployment_filter STRING DEFAULT NULL; WITH latest_state AS ( SELECT flow_run_id, name AS state_name, message, type AS state_type, timestamp, ROW_NUMBER() OVER (PARTITION BY flow_run_id ORDER BY timestamp DESC) AS rn FROM

PROJECT.DATASET.flow_run_state

), base AS ( SELECT m.*, ls.message FROM

PROJECT.DATASET.view_flow_run_metrics_top_level

m JOIN latest_state ls ON ls.flow_run_id = m.id AND ls.rn = 1 WHERE m.end_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL days_lookback DAY) AND ls.state_type IN ('FAILED','CRASHED') AND (deployment_filter IS NULL OR m.deployment_name = deployment_filter) ) SELECT flow_name, deployment_name, message, COUNT(*) AS occurrences FROM base GROUP BY flow_name, deployment_name, message ORDER BY occurrences DESC LIMIT 100;

7) Task-run error rate by flow (tasks under top-level flow runs)

DECLARE days_lookback INT64 DEFAULT 30; DECLARE deployment_filter STRING DEFAULT NULL; WITH tr AS ( SELECT tr.*, fr.flow_id, f.name AS flow_name, d.name AS deployment_name FROM

PROJECT.DATASET.task_run

tr JOIN

PROJECT.DATASET.flow_run

fr ON fr.id = tr.flow_run_id LEFT JOIN

PROJECT.DATASET.flow

f ON f.id = fr.flow_id LEFT JOIN

PROJECT.DATASET.deployment

d ON d.id = fr.deployment_id WHERE fr.parent_task_run_id IS NULL AND tr.end_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL days_lookback DAY) AND (deployment_filter IS NULL OR d.name = deployment_filter) ) SELECT DATE(COALESCE(end_time, start_time, created)) AS run_day, flow_name, COUNT(*) AS task_runs, COUNTIF(state_type IN ('FAILED','CRASHED')) AS failed_task_runs, SAFE_DIVIDE(COUNTIF(state_type IN ('FAILED','CRASHED')), COUNT(*)) AS task_error_rate FROM tr GROUP BY run_day, flow_name ORDER BY run_day, flow_name; ``` Looker/Looker Studio setup (quick start) - Connect to

view_flow_run_metrics_top_level

. - Create a chart for: - Daily success rate by flow: use run_day (dimension), flow_name (breakdown), success_rate = COUNTIF(state_type='COMPLETED') / COUNTIF(state_type IN terminal). Filter control on deployment_name. Set date control to last 7/14/30/90 days. - Daily throughput: runs_completed and runs_started (use separate data sources or blend from queries 3, or create two scorecards by day).

- P95 run duration and queue latency: use query 4 or define fields using quantiles if using LookerML; otherwise pre-aggregate via a scheduled query into a table. - Filters: - deployment_name (dropdown) - Date range (Last 7/14/30/90 days) Want me to: - Swap in your exact PROJECT.DATASET and send these as ready-to-run? - Package this into a Looker Studio template with the filters wired up?

Awesome — here’s a ready-to-build Looker Studio template plan tailored to: - Singular table names - Top-level runs only - Grouped by flow, with a filter on deployment name - Date controls for 7/14/30/90 days It includes the BigQuery objects (1 view + 2 helper objects) and step-by-step report setup. If you share your PROJECT.DATASET, I can swap it in and send you a linkable report stub. Step 0) Replace placeholders - Replace PROJECT.DATASET everywhere below with your values. Step 1) Create a reusable base view for Looker Studio

CREATE OR REPLACE VIEW `PROJECT.DATASET.view_flow_run_metrics_top_level` AS
SELECT
  fr.id,
  fr.flow_id,
  f.name AS flow_name,
  fr.deployment_id,
  d.name AS deployment_name,
  fr.name AS flow_run_name,
  fr.created,
  fr.start_time,
  fr.end_time,
  fr.expected_start_time,
  fr.state_type,
  fr.state_name,
  -- convenience metrics
  IF(fr.start_time IS NOT NULL AND fr.expected_start_time IS NOT NULL,
     TIMESTAMP_DIFF(fr.start_time, fr.expected_start_time, SECOND), NULL) AS queue_latency_s,
  IF(fr.end_time IS NOT NULL AND fr.start_time IS NOT NULL,
     TIMESTAMP_DIFF(fr.end_time, fr.start_time, SECOND), NULL) AS run_duration_s,
  DATE(COALESCE(fr.end_time, fr.start_time, fr.created)) AS run_day
FROM `PROJECT.DATASET.flow_run` fr
LEFT JOIN `PROJECT.DATASET.flow` f ON f.id = fr.flow_id
LEFT JOIN `PROJECT.DATASET.deployment` d ON d.id = fr.deployment_id
WHERE fr.parent_task_run_id IS NULL;

Step 2) Materialize daily percentiles for fast charts (scheduled table) - This avoids percentile math limitations in Looker Studio and makes time-series smooth. - Partition by day for performance.

CREATE OR REPLACE TABLE `PROJECT.DATASET.flow_daily_metrics`
PARTITION BY run_day
CLUSTER BY flow_name, deployment_name AS
SELECT
  run_day,
  flow_id,
  flow_name,
  deployment_name,
  COUNTIF(state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED')) AS terminal_runs,
  SUM(CASE WHEN state_type = 'COMPLETED' THEN 1 ELSE 0 END) AS completed_runs,
  SUM(CASE WHEN state_type IN ('FAILED','CRASHED') THEN 1 ELSE 0 END) AS error_runs,
  SAFE_DIVIDE(SUM(CASE WHEN state_type = 'COMPLETED' THEN 1 ELSE 0 END),
              NULLIF(COUNTIF(state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED')), 0)) AS success_rate,
  SAFE_DIVIDE(SUM(CASE WHEN state_type IN ('FAILED','CRASHED') THEN 1 ELSE 0 END),
              NULLIF(COUNTIF(state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED')), 0)) AS error_rate,
  AVG(run_duration_s) AS avg_run_duration_s,
  APPROX_QUANTILES(run_duration_s, 100)[OFFSET(50)] AS dur_p50_s,
  APPROX_QUANTILES(run_duration_s, 100)[OFFSET(95)] AS dur_p95_s,
  APPROX_QUANTILES(queue_latency_s, 100)[OFFSET(50)] AS ql_p50_s,
  APPROX_QUANTILES(queue_latency_s, 100)[OFFSET(95)] AS ql_p95_s
FROM `PROJECT.DATASET.view_flow_run_metrics_top_level`
WHERE state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED')
GROUP BY run_day, flow_id, flow_name, deployment_name;

- Schedule this query daily (or hourly) in BigQuery (Schedule → Recreate table). - Optional: For a backfill, run once with no WHERE time bound; ongoing schedule can limit to the last 120 days if needed. Step 3) View for failure diagnostics (latest failure/crash message)

CREATE OR REPLACE VIEW `PROJECT.DATASET.view_flow_failed_latest_messages` AS
WITH latest_state AS (
  SELECT
    flow_run_id,
    name AS state_name,
    message,
    type AS state_type,
    timestamp,
    ROW_NUMBER() OVER (PARTITION BY flow_run_id ORDER BY timestamp DESC) AS rn
  FROM `PROJECT.DATASET.flow_run_state`
)
SELECT
  m.run_day,
  m.flow_name,
  m.deployment_name,
  ls.message
FROM `PROJECT.DATASET.view_flow_run_metrics_top_level` m
JOIN latest_state ls
  ON ls.flow_run_id = m.id AND ls.rn = 1
WHERE ls.state_type IN ('FAILED','CRASHED');

Step 4) Add Looker Studio data sources - Source A: BigQuery →

PROJECT.DATASET.view_flow_run_metrics_top_level

- Source B: BigQuery →

PROJECT.DATASET.flow_daily_metrics

- Source C: BigQuery →

PROJECT.DATASET.view_flow_failed_latest_messages

For Source A, add these calculated fields (Data source level) - terminal_numeric:

CASE
  WHEN state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED') THEN 1
  ELSE 0
END

- success_numeric:

CASE WHEN state_type = 'COMPLETED' THEN 1 ELSE 0 END

- error_numeric:

CASE WHEN state_type IN ('FAILED','CRASHED') THEN 1 ELSE 0 END

- run_day_started:

DATE(start_time)

- run_day_completed:

DATE(end_time)

- success_rate:

SAFE_DIVIDE(SUM(success_numeric), NULLIF(SUM(terminal_numeric), 0))

- error_rate:

SAFE_DIVIDE(SUM(error_numeric), NULLIF(SUM(terminal_numeric), 0))

Step 5) Build the Looker Studio report (pages and charts) Page 1: Overview (by flow, filterable by deployment) - Controls: - Date range control (default Last 30 days; provide quick links for 7/14/30/90) - Dropdown control: deployment_name (from Source A or B) - Optional: Dropdown control: flow_name - Scorecards (Source B: flow_daily_metrics; Date range applies to run_day): - Terminal runs: SUM(terminal_runs) - Success rate: AVG(success_rate) (averages daily rate by flow; good for trend; leaderboard uses weighted below) - Error rate: AVG(error_rate) - P95 run duration (s): AVG(dur_p95_s) (approximation across days) - P95 queue latency (s): AVG(ql_p95_s) - Time series: Success rate by day (Source B) - Dimension: run_day - Breakdown: flow_name - Metric: success_rate - Time series: Throughput (two charts, Source A) - Chart 1: Runs started by day - Dimension: run_day_started - Metric: COUNT_DISTINCT(id) - Filter: run_day_started Is not null - Breakdown: flow_name - Chart 2: Runs completed by day (terminal only) - Dimension: run_day_completed - Metric: COUNT_DISTINCT(id) - Filter: state_type IN ('COMPLETED','FAILED','CANCELLED','CRASHED') AND run_day_completed Is not null - Breakdown: flow_name - Table: Flow leaderboard (weighted across the selected period; Source A) - Dimension: flow_name - Metrics: - Runs (terminal): SUM(terminal_numeric) - Completed runs: SUM(success_numeric) - Failed/Crash runs: SUM(error_numeric) - Success rate (weighted): SAFE_DIVIDE(SUM(success_numeric), NULLIF(SUM(terminal_numeric), 0)) - Avg run duration (s): AVG(run_duration_s) (only on runs with duration) - Optional secondary dimension: deployment_name (or add as a filter) Page 2: Latency & Duration details (by flow) - Time series (Source B) - Dimension: run_day - Breakdown: flow_name - Metrics: dur_p50_s, dur_p95_s - Time series (Source B) - Dimension: run_day - Breakdown: flow_name - Metrics: ql_p50_s, ql_p95_s - Table (Source B) - Dimensions: flow_name - Metrics: AVG(dur_p95_s), AVG(ql_p95_s), AVG(avg_run_duration_s), SUM(terminal_runs) - Sort by SUM(terminal_runs) desc Page 3: Failure diagnostics (latest error messages) - Controls: deployment_name, flow_name (optional) - Table (Source C: view_flow_failed_latest_messages) - Dimensions: flow_name, deployment_name, message - Metric: COUNT(*) as occurrences - Sort by occurrences desc - Optional: Time series of failure counts per day - Data source: Source A - Dimension: run_day - Metric: SUM(error_numeric) - Breakdown: flow_name Notes and tips - Date controls: bind to run_day (Sources A, B, C use fields that Looker Studio detects as date). - Deployment filter: place a report-level filter control on deployment_name so all charts respond. - Success/error rates: - Use Source B (daily metrics) for clean per-day time series. - Use Source A (raw runs) for weighted rollups across the date range (leaderboards). - P95 across a long range: - Daily p95 averaged across days is an approximation. If you need exact p95 over the full selected range, I can provide a BigQuery custom-query data source that computes it on the fly, or a scheduled table that aggregates by (flow_name, deployment_name, window).

Want me to generate a shareable Looker Studio report? - If you can share your

PROJECT.DATASET

(read-only is fine) or give me the exact IDs, I’ll: - Create the view/table objects (or send your infra a one-click script) - Stand up a Looker Studio report with the pages and charts above - Return a template link you can copy into your environment If you prefer, send me your actual PROJECT and DATASET and I’ll paste back the SQLs with the placeholders filled so you can run them as-is.