data-engineering-bigquery-gcs

Files (6)

• SKILL.md
• bigquery-advanced-patterns.md · references/
• bigquery-gcs-io.md · references/
• medallion-architecture.md · references/
• repo-evidence.md · references/
• temp-table-merge-pattern.md · references/

SKILL.md (first lines)

# Data Engineering: BigQuery & GCS Pipelines

Stack-agnostic BigQuery data pipeline patterns with medallion layering and Temporal orchestration.

## When to use

- Implementing medallion architecture (bronze/silver/gold layers) for a data warehouse
- Building ETL pipelines that load data to BigQuery
- Designing BigQuery schemas with date partitioning and clustering
- Orchestrating data sync jobs with Temporal workflows and activities
- Implementing atomic upserts (MERGE) from staging tables to production tables
- Loading in-memory pandas DataFrames directly to BigQuery (bypassing GCS staging)
- Implementing CDC deduplication for Datastream-synced bronze tables
- Evolving table schemas dynamically by adding nullable columns
- Executing parameterized BigQuery queries safely without SQL injection
- Streaming small batches of rows into BigQuery tables
- Generating table and column descriptions for data catalogs

## Core conventions

### Medallion Architecture (Bronze → Silver → Gold)

**Three-layer pattern**: Bronze = raw ingestion (Datastream CDC, GCS exports); Silver = cleaned, joined, business logic; Gold = aggregated metrics and KPIs. Each layer is a separate BigQuery dataset.

**Bronze table conventions**: Schema `bronze`, table prefix `br_`, tags `["bronze", "cdc"]`. Use `QUALIFY ROW_NUMBER() OVER (PARTITION BY {pk} ORDER BY created_at DESC, datastream_metadata.source_timestamp DESC) = 1` to deduplicate CDC events. Exclude deletes: `WHERE datastream_metadata.change_type NOT LIKE '%DELETE%'`.

**Silver table conventions**: Schema `silver`, table prefix `slv_`, tags `["silver", "daily"]`. Reference bronze tables via `${ref("br_tablename")}` for Dataform lineage. Silver layer injects `org_id` for multi-tenant analytics.

**Gold table conventions**: Schema `gold`, table prefix `gld_`, tags `["gold", "daily"]`. Reference silver via `${ref("slv_domain_clean")}`. Gold contains aggregations (COUNT, SUM, AVG) grouped by dimensions. If silver already has aggregations, gold passes through `SELECT *`.

**Layer dependencies**: Track via `dependencies: ["br_table1", "br_table2"]` in SQLX config blocks. Each layer references the layer below; never skip layers (bronze → gold is an anti-pattern).

### BigQuery Client Patterns

**Client initialization**: `from google.cloud import bigquery; client = bigquery.Client(project=project_id)`. For explicit project override, pass `project=` to the constructor.

**Query execution**: `rows = client.query(sql_query).result()` for sync execution. Iterate `rows` to consume results. For large result sets, stream rows without loading all into memory.

**Schema introspection**: `table = client.get_table(f"{project}.{dataset}.{table_name}")`. Access `table.schema` for field list, `table.view_query` for view SQL. Each field has `.name`, `.field_type`, `.mode`, `.description`.

**Dry-run for schema inference**: Execute query with `job_config.dry_run = True` to get schema without running query. Use this to derive table schema before first load.

**Table creation with partitioning**: `CREATE TABLE {table} (...) PARTITION BY metric_date CLUSTER BY org_id, site`. Always partition by date column for time-series data; cluster by high-cardinality dimensions used in filters. _(Pattern inferred from BigQuery best practices; not directly observed in evidence repos, which use SQLX/Dataform declarative config instead of raw DDL)_

**Load job from GCS**: Use `client.load_table_from_uri(gcs_pattern, table_ref, job_config=LoadJobConfig(...))` to ingest JSON/CSV/Parquet from GCS into BigQuery. Common config options: `source_format`, `schema`, `write_disposition` ("WRITE_TRUNCATE"|"WRITE_APPEND"), `ignore_unknown_values=True` for schema flexibility, `job_id_prefix` for tracking.

**MERGE for upserts**: Use BigQuery MERGE to atomically update existing rows and insert new ones. Pattern: Load to temp table → MERGE temp to main → cleanup temp. MERGE condition typically checks a timestamp to only update when source is newer.

**Schema building for nested fields**: For RECORD types, use recursive schema building with `bigquery.SchemaField(name, "RECORD", fields=[...])`. Each nested field is itself a SchemaField with name, type, and mode.

> The six patterns below are summarized here; full runnable examples (with variants) live in `references/bigquery-advanced-patterns.md`.

### Parameterized Queries (Safe SQL Injection Prevention)

**Pattern**: Use `QueryJobConfig(query_parameters=[...])` with `@param_name` placeholders in SQL to safely pass user input without string interpolation. Supports scalar and array parameters. BigQuery validates types at execution time.

**ScalarQueryParameter**: For single values. `bigquery.ScalarQueryParameter("run_id", "STRING", run_id)` binds `@run_id` in the query. Common types: `"STRING"`, `"INT64"`, `"FLOAT64"`, `"DATE"`, `"TIMESTAMP"`.

**ArrayQueryParameter**: For list values in `IN` clauses or `UNNEST`. `bigquery.ArrayQueryParameter("table_names", "STRING", ["table1", "table2"])` binds `@table_names`. Use with `WHERE column IN UNNEST(@table_names)`.

### Streaming Inserts (Small-Batch Row Insertion)

**Pattern**: Use `client.insert_rows_json(table_ref, rows_batch)` to stream small batches of rows into BigQuery without load jobs. Ideal for near-real-time ingestion when batch size is <1000 rows. Batch inserts to avoid per-row overhead; use `BATCH_SIZE = 500` as a starting point. Always check for `errors` in the return value.

**Batching logic**: Accumulate rows in a list, flush when batch size is reached or processing is complete. Flush remaining rows at the end.

**Error handling**: `insert_rows_json` returns a list of error dicts for failed rows. Log errors and either retry or skip failed rows. Non-empty errors indicate partial failure.

### load_table_from_dataframe (In-Memory DataFrame Loads)

**Pattern**: Use `client.load_table_from_dataframe(df, table_fqn, job_config=LoadJobConfig(...)).result()` to load pandas DataFrames directly into BigQuery without intermediate GCS staging. Replaces the old pattern of writing to GCS Parquet then loading. Always specify `write_disposition` (`WRITE_APPEND` or `WRITE_TRUNCATE`).

**When to use**: In-memory data transformations (pandas, Excel uploads, API responses) that fit in memory. For >1GB datasets, prefer `load_table_from_uri` from GCS.

**Schema inference**: BigQuery infers schema from DataFrame dtypes. For explicit schema control, pass `job_config.schema=[...]`.

### Dynamic Schema Evolution (Adding NULLABLE Columns)

**Pattern**: Compare incoming data columns against existing table schema, compute missing columns, append them as NULLABLE `SchemaField` objects, and update the table via `client.update_table(table, ["schema"])`. Only NULLABLE columns can be added; NOT NULL columns require a full table rebuild.

**When to use**: Semi-structured data sources (JSON from APIs, evolving GCS exports) where new fields appear over time. Prevents load job failures when upstream schema changes.

### TimePartitioning Python API (Programmatic Partitioning)

**Pattern**: Set `table.time_partitioning = bigquery.TimePartitioning(type_=TimePartitioningType.DAY, field="timestamp_column")` before calling `client.create_table(table)`. For declarative SQLX/Dataform, use `PARTITION BY` in DDL; for programmatic table creation, use the Python API.

**When to use**: Creating tables via Python code (not SQLX). Partitioning reduces query costs by pruning partitions based on date filters. Pair with `table.clustering_fields = [...]` and `expiration_ms` for rolling retention.

### BigQueryUtils Wrapper Class (Reusable Client Pattern)

**Pattern**: Encapsulate BigQuery client initialization and common operations (`query_data`, `load_dataframe`, `insert_rows`) in a reusable class. Centralizes error handling, logging, and project configuration. Each instance manages a single `bigquery.Client` — initialize once and reuse (clients are expensive to construct).

**When to use**: Multiple modules need BigQuery access with consistent error handling and logging, and easier mocking for unit tests. Avoids repeating client initialization and try/except blocks.

### GCS Client Patterns (See gcs-file-storage-patterns Skill)

For GCS blob upload/download operations, refer to the `gcs-file-storage-patterns` skill. The `load_table_from_dataframe` pattern above replaces intermediate GCS Parquet staging for in-memory data; use GCS staging (`load_table_from_uri`) only for >1GB datasets or when data originates in GCS.

### Temporal Orchestration (Activity-Based I/O)

**Critical rule: Heavy I/O in activities, not workflows**: BigQuery queries, GCS uploads/downloads, pandas transforms, and external API calls MUST run inside Temporal activities, never directly in workflow code. Workflows are replay-safe and should only orchestrate, not execute blocking I/O.

**Activity pattern**: Define activity functions decorated with `@activity.defn`. Each activity should be idempotent. Use activities for: BigQuery queries, GCS blob operations, database writes, HTTP requests, pandas processing.

**Workflow pattern**: Workflow functions orchestrate by calling `await workflow.execute_activity(activity_fn, args, ...)`. Workflows maintain state and handle retries, but do not perform I/O.

**Parent-child workflow chaining**: Use `workflow.execute_child_workflow()` to sequentially chain workflows (e.g., a parent workflow chains ManualQueryWorkflow → MaterializeQueryWorkflow → SyncMetadataWorkflow → DataSyncWorkflow). If a child fails, parent can trigger rollback activities.

**Rollback on failure**: When a child workflow fails, parent workflow invokes a rollback activity (e.g., `rollback_by_usecase_activity`) to clean up database records and BigQuery resources by identifier.

### Partitioning and Schema Conventions

**Date partitioning**: All time-series tables MUST use `PARTITION BY {date_column}` (e.g., `metric_date`, `event_date`). This enables efficient pruning and reduces query costs. _(inferred from multi-tenancy org_id pattern)_

**Clustering**: After partitioning, add `CLUSTER BY` for high-cardinality columns used in WHERE clauses and JOINs (e.g., `org_id`, `site`, `product_id`). Clustering is free and improves query performance. _(inferred from multi-tenancy org_id pattern)_

**org_id for multi-tenant data**: Every Silver and Gold table MUST include `org_id STRING NOT NULL` as a mandatory column for multi-org analytics. Cluster by `org_id` first for efficient tenant scoping. _(Pattern inferred from multi-tenancy conventions and medallion generator schema handling)_

**Audit columns**: Add `_ingested_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP()` to Bronze and Silver tables. For CDC bronze, use `datastream_metadata.source_timestamp` for source change tracking. _(inferred from bronze_dataset_generator.py deduplication logic)_

From ajyadav013/claude-kit, MIT.

More from ajyadav013/claude-kit