open-navigator / web_docs /docs /dbt /migration-python-to-dbt.md
jcbowyer's picture
Clean HuggingFace deployment without binary files
e59d91d
|
Raw
History Blame Contribute Delete
6.9 kB
metadata
sidebar_position: 5

Migrating JSONB Extraction from Python to dbt

Overview

This guide shows how to replace scripts/datasources/gemini/load_meeting_transcripts_bronze.py with dbt incremental models.

Why Migrate to dbt?

Current approach (Python):

  • ❌ Separate tool/language from main transformation pipeline
  • ❌ Manual UPSERT logic with psycopg2
  • ❌ Harder to track lineage
  • ❌ No built-in testing framework

dbt approach:

  • βœ… Single tool for all transformations
  • βœ… SQL-based JSONB extraction (simpler to maintain)
  • βœ… Incremental processing built-in
  • βœ… Automatic lineage tracking
  • βœ… Built-in testing and documentation
  • βœ… Version control friendly

What CAN Be Migrated

βœ… load_meeting_transcripts_bronze.py - JSONB extraction to bronze tables

  • bronze_contacts
  • bronze_decisions
  • bronze_topics
  • bronze_organizations_meetings
  • bronze_bills
  • bronze_causes
  • bronze_financial_items

What CANNOT Be Migrated

❌ load_meeting_transcripts.py - Calls Gemini API (data loading, not transformation) ❌ check_models_used.py - Reporting script ❌ analyze_with_multi_models.py - Calls Gemini API ❌ migrations/cleanup_null_records.py - One-time migration

Migration Steps

1. Create dbt Bronze Models

Created example models (see models/bronze/):

  • bronze_contacts_from_ai.sql - Extract people from JSONB
  • bronze_decisions_from_ai.sql - Extract decisions from JSONB

Pattern for other tables:

{{
  config(
    materialized='incremental',
    unique_key='source_event_id_<entity>_id',
    schema='bronze',
    tags=['bronze', 'incremental', 'ai-extraction']
  )
}}

WITH source_events AS (
    SELECT 
        id as event_id,
        structured_analysis,
        ai_model,
        created_at
    FROM {{ source('bronze', 'bronze_events_analysis_ai') }}
    WHERE structured_analysis IS NOT NULL
    
    {% if is_incremental() %}
        AND created_at > (SELECT MAX(extracted_at) FROM {{ this }})
    {% endif %}
),

entities_unnested AS (
    SELECT 
        event_id,
        ai_model,
        jsonb_array_elements(structured_analysis->'<entity_key>') as entity_data,
        created_at as extracted_at
    FROM source_events
    WHERE structured_analysis ? '<entity_key>'
)

SELECT
    -- Extract fields from JSONB...
FROM entities_unnested

2. Update Staging Models

OLD (reads from bronze tables created by Python):

SELECT * FROM {{ source('bronze', 'bronze_contacts') }}

NEW (reads from dbt bronze models):

SELECT * FROM {{ ref('bronze_contacts_from_ai') }}

3. Update Workflow

OLD:

# Step 1: Run Gemini API analysis
python scripts/datasources/gemini/load_meeting_transcripts.py

# Step 2: Extract JSONB to bronze tables (Python)
python scripts/datasources/gemini/load_meeting_transcripts_bronze.py

# Step 3: Transform with dbt
cd dbt_project
dbt run --select stg_bronze_contacts+

NEW:

# Step 1: Run Gemini API analysis (still Python)
python scripts/datasources/gemini/load_meeting_transcripts.py

# Step 2: Extract JSONB with dbt (incremental)
cd dbt_project
dbt run --select bronze_contacts_from_ai+

# Or run all bronze extractions:
dbt run --select tag:ai-extraction+

4. First-Time Setup

Initial load (one time):

# Let Python script create initial bronze tables
python scripts/datasources/gemini/load_meeting_transcripts_bronze.py

# OR let dbt create them fresh:
cd dbt_project
dbt run --select tag:ai-extraction --full-refresh

Going forward (incremental):

cd dbt_project
dbt run --select tag:ai-extraction
# Only processes new records added since last run

JSONB Extraction Patterns

Simple Array Extraction

-- Extract people array
jsonb_array_elements(structured_analysis->'people') as person_data

-- Get fields
person_data->>'full_name' as full_name  -- Text
person_data->>'is_lobbyist' as is_lobbyist  -- Still text, cast later
(person_data->>'is_lobbyist')::boolean  -- Cast to boolean

Nested JSONB (Keep as JSONB)

-- Keep complex structures as JSONB
person_data->'lobbyist_clients' as lobbyist_clients,  -- JSONB
decision_data->'vote_tally' as vote_tally,  -- JSONB
decision_data->'frame_analysis' as frame_analysis  -- JSONB

Date/Numeric Casting

(decision_data->>'decision_date')::date as decision_date,
(decision_data->>'year')::integer as year,
(financial_data->>'amount')::numeric as amount

Handling NULL/Missing Fields

-- Use COALESCE for defaults
COALESCE((person_data->>'is_lobbyist')::boolean, FALSE) as is_lobbyist,

-- Check if key exists
WHERE structured_analysis ? 'people'

Incremental Model Strategy

dbt will:

  1. Check if target table exists
  2. If first run β†’ create full table
  3. If subsequent run β†’ only process new source records
  4. Use unique_key to deduplicate (UPSERT behavior)
{% if is_incremental() %}
    AND created_at > (SELECT MAX(extracted_at) FROM {{ this }})
{% endif %}

Testing

Add tests in _bronze.yml:

models:
  - name: bronze_contacts_from_ai
    description: "Contacts extracted from Gemini AI analysis JSONB"
    tests:
      - dbt_utils.recency:
          datepart: day
          field: extracted_at
          interval: 7
    columns:
      - name: person_id
        tests:
          - not_null
      - name: full_name
        tests:
          - not_null

Performance

Python script:

  • Loads entire events_text_ai table into memory
  • Processes all rows every run
  • Manual batching required

dbt incremental:

  • Only processes new records
  • SQL-based (database does the work)
  • No memory constraints
  • Parallel execution possible

Rollback Plan

If needed, can run both in parallel:

# Old way (Python)
python scripts/datasources/gemini/load_meeting_transcripts_bronze.py \
  --table bronze_contacts_python

# New way (dbt)
dbt run --select bronze_contacts_from_ai

# Compare results
psql -c "SELECT COUNT(*) FROM bronze.bronze_contacts_python"
psql -c "SELECT COUNT(*) FROM bronze.bronze_contacts_from_ai"

Next Steps

  1. βœ… Review example models (bronze_contacts_from_ai.sql, bronze_decisions_from_ai.sql)
  2. Create remaining bronze models:
    • bronze_topics_from_ai.sql
    • bronze_organizations_meetings_from_ai.sql
    • bronze_bills_from_ai.sql
    • bronze_causes_from_ai.sql
    • bronze_financial_items_from_ai.sql
  3. Update staging models to reference new dbt models
  4. Test incremental runs
  5. Deprecate Python extraction script
  6. Update documentation

Questions?

  • How do I handle deduplication? β†’ Use unique_key config
  • What if JSONB structure changes? β†’ Update the model SQL
  • Can I still use Python for initial load? β†’ Yes, for backward compatibility
  • How do I backfill? β†’ Use dbt run --select <model> --full-refresh