Working with Large-Scale Datasets Questions

Brief comparison (Live / DirectQuery vs Extract / Import):

• Data freshness:

  • Live: real-time or near-real-time—queries reflect latest source data.
  • Extract: freshness depends on refresh schedule (hourly, nightly); not real-time.

• Query load on source:

  • Live: every user interaction issues queries to the source → high load with many concurrent users.
  • Extract: source hit limited to refresh windows; users query the BI engine instead → much lower operational load.

• Initial load time:

  • Live: dashboard renders once source responds; initial renders can be fast if source indexed, but can be slow for complex joins/aggregations.
  • Extract: initial extract can be heavy (full load), but subsequent loads are fast from the in-memory/columnar cache.

• Interactivity:

  • Live: can be interactive if the source supports fast ad-hoc queries and concurrency; otherwise sluggish.
  • Extract: highly interactive (in-memory engine, client-side query acceleration, caching) even under many concurrent users.

• Caching:

  • Live: depends on source and BI tool query caching; cache lifetimes are short if data freshness required.
  • Extract: built-in cache (Power BI VertiPaq, Tableau Hyper) and query result caching produce consistent fast UX.

Recommended strategy for an operational KPI dashboard updated hourly with many concurrent users:

1. Use extracts/imports as primary approach:

   • Schedule incremental refresh every hour (or use CDC) to minimize extract time and keep data within SLA.
   • Use partitioning/incremental loads to avoid full refreshes.

2. Design for performance:

   • Build aggregated materialized tables in the warehouse for dashboard-level KPIs (pre-aggregate by hour, region).
   • Push heavy filtering/joins to the warehouse (query folding) so extracts are compact.

3. Hybrid options for specific needs:

   • If a single metric truly needs near-real-time (seconds), keep that metric via a lightweight live connection to a dedicated, scaled source (read-replica) and combine into the dashboard using blended data or composite models (Power BI DirectQuery + Import).

4. Infrastructure & BI settings:

   • Use read replicas or a reporting mart to isolate reporting load.
   • Enable BI tool caching, tune cache TTL to just under refresh interval.
   • Limit complex visuals, use fewer quick filters, and ensure visuals use indexed fields.

5. Concurrency considerations:

   • Test with expected concurrent sessions; scale gateway/refresh capacity.
   • Monitor refresh durations and query times; raise refresh window or further aggregate if needed.

Why this approach: extracts minimize query load and maximize interactivity for many users; hourly incremental refresh meets the stated freshness requirement. Hybrid/live only where true real-time is required, and always isolate reporting from OLTP to protect source performance.

When prototyping a metric on a 200M-row transactions table, sampling speeds iteration. Below are three practical strategies, SQL snippets for a typical warehouse (BigQuery/Postgres-compatible), and bias considerations.

1. Uniform random sampling

• Approach: draw a random subset across all rows to approximate overall distribution.
• SQL:

SELECT *
FROM transactions
WHERE FARM_FINGERPRINT(CAST(transaction_id AS STRING)) % 1000 = 1;
-- ~0.1% sample (1/1000); use RAND() or TABLESAMPLE where supported

• Bias: unbiased for global aggregates if sample size large enough, but rare events or small subgroups may be underrepresented.

2. Stratified sampling by user cohort

• Approach: ensure proportional (or equal) representation across cohorts (e.g., signup_month or customer_tier).
• SQL:

WITH cohorts AS (
  SELECT user_id, signup_month FROM users
)
SELECT t.*
FROM transactions t
JOIN cohorts c USING (user_id)
WHERE MOD(FARM_FINGERPRINT(CONCAT(CAST(user_id AS STRING), signup_month)), 100) < 1;
-- samples ~1% within each cohort key

• Bias: reduces variance across cohorts and prevents dominance by large cohorts. Risk if cohort key is mis-specified or cohorts themselves are sparse.

3. Time-based sampling for seasonality

• Approach: sample full-day or full-week slices across periods (e.g., pick specific days each month) to capture temporal patterns.
• SQL:

SELECT *
FROM transactions
WHERE DATE(created_at) IN (
  SELECT DATE_TRUNC('month', created_at) + INTERVAL '7 days' FROM UNNEST(GENERATE_DATE_ARRAY('2024-01-01','2024-12-01','1 month')) AS d
);
-- picks the 7th day of each month

• Bias: preserves intra-day/day-of-week patterns for chosen slices but may miss irregular events; choose multiple days or rotate to reduce bias.

Practical tips: validate sample estimates against a larger sample or full-table aggregation for a few key metrics; increase sample size for segments with high variance; document sampling method so stakeholders understand limitations.

Approach summary:

• Exact: use DISTINCT count over a 30-day sliding window per day — accurate but expensive for billions of rows.
• Approximate: use HyperLogLog (HLL) sketches for per-day aggregation, then merge sketches for each 30-day window — far cheaper and scalable with bounded memory and fast merges. Trade-off: small, tunable error (~1–2%).

Exact (conceptual, heavy):

-- may be very slow / expensive at scale
SELECT
  d.day,
  COUNT(DISTINCT e.user_id) AS active_30d
FROM
  (SELECT DISTINCT event_date AS day FROM events) d
LEFT JOIN events e
  ON e.event_date BETWEEN DATE_SUB(d.day, INTERVAL 29 DAY) AND d.day
GROUP BY d.day
ORDER BY d.day;

Approximate with HLL (BigQuery):

-- per-day sketch
CREATE TABLE daily_hll AS
SELECT
  event_date,
  HLL_COUNT.INIT(user_id) AS hll
FROM events
GROUP BY event_date;

-- rolling 30-day merge
SELECT
  d.event_date AS day,
  HLL_COUNT.MERGE_AGG(hll) AS hll_30d,
  HLL_COUNT.EXTRACT(HLL_COUNT.MERGE_AGG(hll)) AS approx_active_30d
FROM daily_hll d
JOIN daily_hll window
  ON window.event_date BETWEEN DATE_SUB(d.event_date, INTERVAL 29 DAY) AND d.event_date
GROUP BY d.event_date
ORDER BY d.event_date;

Approximate with HLL (Snowflake using HLL UDFs or APPROX_COUNT_DISTINCT):

-- Snowflake has APPROX_COUNT_DISTINCT and HLL functions via extensions
SELECT
  day,
  APPROX_COUNT_DISTINCT(user_id, 0.02) OVER (ORDER BY day
    RANGE BETWEEN INTERVAL '29' DAY PRECEDING AND CURRENT ROW) AS approx_active_30d
FROM (
  SELECT event_date AS day, user_id FROM events
);

Key trade-offs and implementation notes:

• Accuracy vs cost: HLL gives fixed-memory, fast merges; error controlled by sketch size (e.g., 1–2% typical). Exact distinct is precise but often infeasible at billions of rows.
• Storage pattern: precompute daily HLL sketches (or daily distinct user lists) and materialize them (partitioned by date). Then rolling windows merge only 30 rows per date — cheap.
• Latency: Materialized daily sketches allow fast dashboard refreshes; recompute daily.
• Edge cases: timezone normalization, late-arriving events (use event ingestion windows), and retention.
• Validation: periodically compare approximate vs exact on sampled ranges to monitor error and adjust sketch precision.

Approach summary: exact SQL for correctness when data volume allows; approximate sketches (t-digest, DDSketch, CKMS, quantile digest) for scale/low-memory. Store per-endpoint-per-day aggregates (sketch blobs + counts) and merge at query time.

Exact (batch / single-node SQL)

• Use PERCENTILE_CONT for continuous percentile on aggregated per-day data (works when scanning full rows or pre-aggregates):

-- Postgres/Redshift style: compute daily 95th per endpoint
SELECT endpoint, day,
  PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY latency_ms) AS p95_ms
FROM raw_logs
WHERE day BETWEEN '2025-11-01' AND '2025-11-30'
GROUP BY endpoint, day;

• If raw_rows=500M and too slow, pre-aggregate latencies into buckets or store sampled subsets.

Approximate (recommended at scale)

1. t-digest (good for tail quantiles, mergeable)

• Build t-digest per endpoint/day in an ETL job (Spark, Python). Persist serialized digest (binary or Base64) and row_count, min, max.
• Merge digests at query time by loading and combining digests.

Python example (tdigest):

from tdigest import TDigest
# build per-partition
td = TDigest()
for x in latencies:
    td.update(x)
blob = td.compress().serialize()  # store blob in table

Store schema:

• endpoint (string), day (date), digest_blob (bytes), count (int), min_ms, max_ms

Query pattern (merge server-side or in-engine):

• If DB supports UDFs, implement merge_tdigest(blob[]) → blob, then quantile_udf(blob, 0.95) → float.
• In Spark/Presto: aggregate by merging TDigest objects and then extract quantile.

SQL-like pattern (pseudo):

SELECT endpoint, day,
  quantile_from_merged_tdigest(collect(digest_blob), 0.95) AS p95
FROM endpoint_day_digests
GROUP BY endpoint, day;

2. DDSketch / HLL-like quantile sketches

• DDSketch preserves relative error for large values (good for latency). Same storage/merge pattern: serialized sketch per partition, merge via library UDF.

3. Histogram-based (fixed buckets)

• Pre-aggregate counts per bucket per endpoint/day. To compute 95th, accumulate cumulative counts until 95% point and linearly interpolate within bucket. Simpler, smaller storage, less accurate at tail.

Merging and storage details

• Store one row per endpoint/day per ingestion shard or per worker; design primary key (endpoint, day, shard_id). Columns: sketch_blob, count, min, max, bucket_counts (optional).
• Merging: use library-provided merge function (idempotent). For SQL engines without native merge, perform merge in ETL (Spark/Flink) and write final per-endpoint-day sketch table.

Query patterns for dashboards (fast)

• Precompute daily p95 from sketches nightly and store numeric p95 in metrics table for dashboard reads (fast path).
• For ad-hoc queries over ranges, either:
  • Merge sketches on-the-fly (cost proportional to number of days/endpoints), or
  • Pre-aggregate weekly/monthly sketches to reduce merges.

Trade-offs

• Accuracy: exact PERCENTILE_CONT = exact (subject to sort/float). t-digest/DDSketch offer configurable error (e.g., 1-2% relative) and are very accurate for tails with proper compression.
• Storage: sketches ~1–10KB per (endpoint,day); histograms can be smaller (tens to hundreds bytes) depending on buckets. Storing raw rows far larger.
• Compute: exact needs full data scan and sort (heavy). Sketches require one-pass aggregation and cheap merges; query-time extraction is O(k) where k is sketch size.
• Complexity: implementing sketches needs libraries/UDFs + validation. Histograms easier but less tail accuracy.
• Freshness: real-time streaming with sketches possible (incremental), exact requires streaming materialized views or heavy compute.

Recommended pattern for BI role

• Produce nightly precomputed p95 using sketches (tdigest/DDSketch) + store numeric p95 in a metrics table for dashboards.
• Keep raw sketches in a compact table for ad-hoc re-computation or audits (store blobs, counts, min/max).
• Monitor error: sample exact percentiles periodically on smaller slices to validate sketch fidelity.

Requirements & constraints:

• Ingest ~1B events/day, produce daily aggregated metrics for dashboards (hourly/near-real-time optional), durable, accurate, traceable, low operational overhead for BI owners. Target SLA: daily refresh within 2–4 hours; near-real-time optional for subset.

High-level architecture:

• Raw ingestion → landing in cloud object storage (e.g., S3/GCS) as partitioned, compressed Parquet/ORC files → message bus (Kafka/PubSub) for streaming needs → ETL compute (Spark/Databricks or BigQuery/Athena Redshift Spectrum) → curated analytics tables / materialized aggregate tables → BI layer (Looker/Tableau).

Partitioning & storage:

• Store raw events in time-partitioned folders: dt=YYYY-MM-DD/hour=HH and use larger file sizes (100–512MB) and sorted by high-cardinality keys (e.g., user_id) to improve scan perf.
• Columnar Parquet/ORC with partition pruning and Z-ordering (if supported) for common query patterns.
• Aggregates stored in partitioned fact tables (daily/hourly) in a warehouse (BigQuery/Redshift/Snowflake) with clustering on dimensions used by dashboards.

Incremental processing:

• Use micro-batch or daily batch that processes only new partitions (incremental by dt/hour). Maintain a job-run ledger (metadata table) recording input partitions processed, offsets, and checksum.
• For streaming near-real-time aggregates, use stream processing (Spark Structured Streaming / Flink / kstreams) that writes to materialized view tables.

Idempotency & deduplication:

• Produce deterministic keys for events (hash of sourceID + timestamp + eventType) on ingest. Use those keys in ETL to dedupe using dedupe step: keep latest or first occurrence depending on business rule.
• Write aggregates via upsert / MERGE into warehouse keyed by partition + aggregate key. Ensure ETL uses transactional/atomic writes where supported (BigQuery write dispositions, Snowflake transactions).
• Use dedupe windows for streaming (e.g., 24–48h) to capture late arrivals.

Schema evolution:

• Enforce a schema registry for event contracts (protobuf/avro/JSON schema). ETL reads schema metadata and handles:
  • Additive changes: auto-include new nullable columns.
  • Breaking changes: flag via schema compatibility checks; route to a staging area and alert owners for mapping.
• Use column mapping with default/nulls and transformation logic versioned in code.

Data validation tests:

• Pre- and post-ETL checks:
  • Row count delta vs expected, null-rate thresholds, distinct count checks for keys.
  • Referential integrity for joins (orphan rates).
  • Value-range and spike detection using historical baselines and statistical tests.
• Implement unit tests for transformations and data quality rules in CI (Great Expectations or Deequ).

Alerting & monitoring:

• Pipeline monitoring: job success/failures, SLA lateness, data-quality rule breaches.
• Alerts via PagerDuty/Slack/email with runbook and sample failing partitions. Store audit logs with pointers to sample raw files and REPRO steps.
• Maintain dashboards for pipeline health (throughput, latency, error rates).

Acceptable latency trade-offs:

• Prioritize correctness for daily aggregates; 2–4h latency acceptable. For real-time KPIs, expose lightweight streaming aggregates for small dimension sets with higher cost.
• Trade storage & compute: more frequent processing reduces staleness but increases cost. Use incremental processing and partition pruning to minimize compute.

Example operational pattern:

• Nightly job picks dt partitions, validates, dedupes, computes daily aggregates, MERGE into warehouse; if DQ fails, mark dataset as quarantined and alert; BI dashboards show last successful run timestamp and data freshness.

This design balances scalability, accuracy, and maintainability for BI consumers while enabling optional low-latency views for critical metrics.