File size: 6,903 Bytes
e59d91d
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
---
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:
```sql
{{
  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):**
```sql
SELECT * FROM {{ source('bronze', 'bronze_contacts') }}
```

**NEW (reads from dbt bronze models):**
```sql
SELECT * FROM {{ ref('bronze_contacts_from_ai') }}
```

### 3. Update Workflow

**OLD:**
```bash
# 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:**
```bash
# 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):**
```bash
# 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):**
```bash
cd dbt_project
dbt run --select tag:ai-extraction
# Only processes new records added since last run
```

## JSONB Extraction Patterns

### Simple Array Extraction

```sql
-- 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)

```sql
-- 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

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

### Handling NULL/Missing Fields

```sql
-- 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)

```sql
{% if is_incremental() %}
    AND created_at > (SELECT MAX(extracted_at) FROM {{ this }})
{% endif %}
```

## Testing

Add tests in `_bronze.yml`:

```yaml
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:
```bash
# 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`