# Data Loading

> A Claude skill from nguyenthanhtat/screen1-claude.

- **Type:** Skill
- **Install:** `agentstack add skill-nguyenthanhtat-screen1-claude-data-loading`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [nguyenthanhtat](https://agentstack.voostack.com/s/nguyenthanhtat)
- **Installs:** 0
- **Category:** [Agent Skills](https://agentstack.voostack.com/c/agent-skills)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [nguyenthanhtat](https://github.com/nguyenthanhtat)
- **Source:** https://github.com/nguyenthanhtat/screen1-claude/tree/main/skills/bigquery/data-loading

## Install

```sh
agentstack add skill-nguyenthanhtat-screen1-claude-data-loading
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# BigQuery Data Loading & Export

**Parent Skill:** `/bigquery`  
**Path:** `/bigquery/data-loading`

## Purpose
Efficiently import data from various sources (GCS, local files, streaming) and export BigQuery results.

## When to Use

**Trigger automatically when:**
- Keywords: load, import, export, CSV, JSON, Parquet, streaming
- User uploads file and mentions BigQuery
- User asks to "save query results"

**Chat commands:**
```bash
/bigquery/data-loading import CSV from gs://bucket/data.csv
/bigquery/data-loading load this file into table events
/bigquery/data-loading export query results to GCS
/bigquery/data-loading setup streaming for real-time data
```

## Requirements

- BigQuery access
- Cloud Storage bucket (for GCS loads)
- Appropriate IAM roles:
  - bigquery.tables.create
  - bigquery.tables.updateData
  - storage.objects.get (for GCS)

## Verification
```bash
# Check GCS access
gsutil ls gs://your-bucket/

# Check BigQuery permissions
bq show your_project:dataset
```

---

## Loading Patterns

### Pattern 1: Load CSV from GCS

```bash
# Using bq CLI
bq load \
  --source_format=CSV \
  --skip_leading_rows=1 \
  --autodetect \
  dataset.table_name \
  gs://bucket/data.csv
```

```python
# Using Python client
from google.cloud import bigquery

client = bigquery.Client()

job_config = bigquery.LoadJobConfig(
    source_format=bigquery.SourceFormat.CSV,
    skip_leading_rows=1,
    autodetect=True,
    write_disposition='WRITE_TRUNCATE'  # or WRITE_APPEND
)

uri = "gs://bucket/data.csv"
table_id = "project.dataset.table"

load_job = client.load_table_from_uri(
    uri, table_id, job_config=job_config
)

load_job.result()  # Wait for completion
print(f"Loaded {load_job.output_rows} rows")
```

**Best Practices:**
- Use `--autodetect` for schema inference (development only)
- Specify explicit schema for production
- Use `WRITE_APPEND` for incremental loads
- Partition target table by date if loading daily

---

### Pattern 2: Load JSON (Newline-Delimited)

```python
job_config = bigquery.LoadJobConfig(
    source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON,
    autodetect=True,
)

uri = "gs://bucket/events.json"
load_job = client.load_table_from_uri(uri, table_id, job_config=job_config)
load_job.result()
```

**JSON Format Required:**
```json
{"user_id": "123", "event": "click", "timestamp": "2024-02-09T10:00:00Z"}
{"user_id": "456", "event": "view", "timestamp": "2024-02-09T10:01:00Z"}
```

**NOT standard JSON array:**
```json
// ❌ This won't work
[
  {"user_id": "123", ...},
  {"user_id": "456", ...}
]
```

---

### Pattern 3: Load Parquet (Recommended for Large Data)

```python
job_config = bigquery.LoadJobConfig(
    source_format=bigquery.SourceFormat.PARQUET,
    write_disposition='WRITE_TRUNCATE'
)

uri = "gs://bucket/data.parquet"
load_job = client.load_table_from_uri(uri, table_id, job_config=job_config)
load_job.result()
```

**Advantages:**
- Compressed (smaller than CSV)
- Schema included (no need for autodetect)
- Faster load times (5-10x vs CSV)
- Columnar format = better query performance

---

### Pattern 4: Streaming Inserts (Real-Time)

```python
# For real-time data (e.g., website events)
rows_to_insert = [
    {"user_id": "123", "event": "purchase", "timestamp": "2024-02-09T10:00:00"},
    {"user_id": "456", "event": "view", "timestamp": "2024-02-09T10:01:00"},
]

errors = client.insert_rows_json(table_id, rows_to_insert)

if errors:
    print(f"Errors: {errors}")
else:
    print("Rows inserted successfully")
```

**Important:**
- Streaming has cost: $0.01 per 200MB
- Use only for real-time requirements
- For batch: use load jobs instead
- Has ~90 second buffer before data is queryable
- Max 10,000 rows per request

---

### Pattern 5: Load from Local File

```python
# Upload local CSV to BigQuery
with open("local_data.csv", "rb") as source_file:
    job = client.load_table_from_file(
        source_file,
        table_id,
        job_config=job_config
    )

job.result()
print(f"Loaded {job.output_rows} rows")
```

**Use cases:**
- Small files ( 1GB results, use wildcards for automatic sharding
bq extract \
  --destination_format=CSV \
  dataset.large_table \
  gs://bucket/shard-*.csv

# Creates: shard-000000000000.csv, shard-000000000001.csv, ...
```

**BigQuery automatically shards files > 1GB**

---

### Export 3: Export Query Results Directly

```python
# Export query result without creating intermediate table
query = """
SELECT 
  user_id,
  COUNT(*) as event_count
FROM events
WHERE DATE(timestamp) = CURRENT_DATE() - 1
GROUP BY user_id
"""

# Configure export
job_config = bigquery.QueryJobConfig(
    destination="project.dataset.temp_export"
)

# Run query
query_job = client.query(query, job_config=job_config)
query_job.result()

# Export the result table
destination_uri = "gs://bucket/export.csv"
extract_job = client.extract_table(
    "project.dataset.temp_export",
    destination_uri
)
extract_job.result()

# Clean up temp table
client.delete_table("project.dataset.temp_export")
```

---

## Schema Management

### Explicit Schema (Production Recommended)

```python
from google.cloud import bigquery

schema = [
    bigquery.SchemaField("user_id", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("timestamp", "TIMESTAMP", mode="REQUIRED"),
    bigquery.SchemaField("event_name", "STRING"),
    bigquery.SchemaField("properties", "JSON"),
    bigquery.SchemaField("metadata", "RECORD", fields=[
        bigquery.SchemaField("source", "STRING"),
        bigquery.SchemaField("version", "INTEGER")
    ])
]

job_config = bigquery.LoadJobConfig(
    schema=schema,
    write_disposition='WRITE_APPEND'
)
```

### Schema Auto-Detection (Development Only)

```python
job_config = bigquery.LoadJobConfig(
    autodetect=True,
    write_disposition='WRITE_TRUNCATE'
)
```

**Warning:** Auto-detect can infer wrong types
- ZIP codes as INTEGER (should be STRING)
- Phone numbers as INTEGER
- Dates as STRING

---

## Incremental Loading Strategy

### Pattern: Daily Incremental Loads

```python
from datetime import datetime, timedelta

# Get yesterday's date
yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')

# Load with partition decorator
partition_date = yesterday.replace('-', '')  # Format: 20240209
table_id = f"project.dataset.events${partition_date}"

# Load data for yesterday only
uri = f"gs://bucket/data-{yesterday}.csv"
load_job = client.load_table_from_uri(uri, table_id, job_config=job_config)
load_job.result()

print(f"Loaded data for partition {yesterday}")
```

### Pattern: Merge/Upsert Strategy

```sql
-- Use MERGE for deduplication and updates
MERGE dataset.target T
USING dataset.staging S
ON T.user_id = S.user_id AND T.date = S.date
WHEN MATCHED THEN
  UPDATE SET 
    event_count = S.event_count,
    last_updated = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN
  INSERT (user_id, date, event_count, last_updated)
  VALUES (S.user_id, S.date, S.event_count, CURRENT_TIMESTAMP());
```

---

## Error Handling

```python
try:
    load_job = client.load_table_from_uri(uri, table_id, job_config=job_config)
    load_job.result()  # Wait and check for errors
    
    print(f"✅ Loaded {load_job.output_rows} rows")
    
except Exception as e:
    print(f"❌ Load failed: {e}")
    
    # Check detailed errors
    if hasattr(load_job, 'errors') and load_job.errors:
        for error in load_job.errors:
            print(f"Error: {error.get('message')}")
            print(f"Location: {error.get('location')}")
            print(f"Reason: {error.get('reason')}")
```

**Common Errors:**

| Error | Cause | Solution |
|-------|-------|----------|
| Schema mismatch | Column types don't match | Use explicit schema |
| Permission denied | Missing IAM roles | Add bigquery.tables.updateData |
| File not found | Wrong GCS path | Check URI with `gsutil ls` |
| Invalid format | Malformed CSV/JSON | Validate file format |
| Quota exceeded | Too many concurrent loads | Wait or request quota increase |

---

## Performance Tips

### Optimize Load Speed

```python
# 1. Use Parquet instead of CSV (5-10x faster)
source_format=bigquery.SourceFormat.PARQUET

# 2. Load multiple files in parallel
uris = [
    "gs://bucket/file1.csv",
    "gs://bucket/file2.csv",
    "gs://bucket/file3.csv"
]
load_job = client.load_table_from_uri(uris, table_id, job_config=job_config)

# 3. Use compression
job_config.compression = 'GZIP'

# 4. Allow some bad records (for dirty data)
job_config.max_bad_records = 100

# 5. Batch loads vs streaming
# Batch: Free, use for scheduled ETL
# Streaming: $0.01/200MB, use only for real-time
```

### Cost Optimization

**Costs:**
- **Batch loads:** FREE (only pay for storage)
- **Streaming:** $0.01 per 200MB
- **Export:** $0.011 per GB (to GCS)
- **Storage:** $0.02/GB/month (active), $0.01/GB/month (long-term)

**Recommendations:**
- Use batch loads whenever possible
- Compress files before loading
- Use Parquet for large datasets
- Partition tables for faster queries

---

## Vietnamese Data Handling

```python
# Ensure UTF-8 encoding for Vietnamese characters
job_config.encoding = 'UTF-8'

# Example CSV with Vietnamese
# user_id,name,location,product
# 1,Nguyễn Văn A,Hồ Chí Minh,Xe ô tô Toyota
# 2,Trần Thị B,Hà Nội,Xe máy Honda
```

**Test Vietnamese characters:**
```python
# After loading
query = """
SELECT name, location
FROM dataset.table
WHERE location LIKE '%Hồ Chí Minh%'
"""
results = client.query(query).result()
for row in results:
    print(f"{row.name} - {row.location}")
```

---

## Integration Examples

### Next.js API Route (Load Data)

```javascript
// pages/api/upload-data.js
import { BigQuery } from '@google-cloud/bigquery';
import { Storage } from '@google-cloud/storage';

export default async function handler(req, res) {
  if (req.method !== 'POST') {
    return res.status(405).json({ error: 'Method not allowed' });
  }

  const bigquery = new BigQuery();
  const storage = new Storage();
  
  try {
    // 1. Upload file to GCS
    const bucket = storage.bucket('my-bucket');
    const fileName = `uploads/${Date.now()}.csv`;
    const file = bucket.file(fileName);
    
    await file.save(req.body);
    
    // 2. Load to BigQuery
    const [job] = await bigquery
      .dataset('my_dataset')
      .table('events')
      .load(file, {
        sourceFormat: 'CSV',
        skipLeadingRows: 1,
        autodetect: true
      });
    
    // 3. Wait for completion
    const [metadata] = await job.getMetadata();
    
    res.json({ 
      success: true, 
      rowsLoaded: metadata.statistics.load.outputRows 
    });
    
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
}
```

---

## Integration with Other Skills

**Workflow:**
1. Load data → `/bigquery/data-loading`
2. Design schema → `/bigquery/schema-design`
3. Optimize queries → `/bigquery/query-optimization`
4. Schedule incremental loads → `/bigquery/scheduled-queries`
5. Monitor costs → `/bigquery/cost-monitoring`

---

## Quick Reference

| Task | Format | Command |
|------|--------|---------|
| Load CSV | CSV | `bq load --source_format=CSV` |
| Real-time | Streaming | `client.insert_rows_json()` |
| One-time query | Federated | `CREATE EXTERNAL TABLE` |
| Export results | Parquet | `bq extract --destination_format=PARQUET` |
| Daily batch | Parquet | Cloud Scheduler + load job |
| Large files | Parquet/Avro | Load from GCS |

## Version

- **Version:** 1.0.0
- **Last Updated:** 2024-02-09

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [nguyenthanhtat](https://github.com/nguyenthanhtat)
- **Source:** [nguyenthanhtat/screen1-claude](https://github.com/nguyenthanhtat/screen1-claude)
- **License:** MIT

Install and usage instructions live in the source repository linked above.

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** no
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/skill-nguyenthanhtat-screen1-claude-data-loading
- Seller: https://agentstack.voostack.com/s/nguyenthanhtat
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.