23 - Data Ingestion — SQL Server, BigQuery, Firestore

47 min read

23. Data Ingestion — SQL Server, BigQuery, Firestore

Quote

“Data is a precious thing and will last longer than the systems themselves.”

— Tim Berners-Lee, attributed remark (c. 2006)


flowchart LR
    LOC["Local Files<br/>CSV · JSON · Parquet"]
    GCS["GCS<br/>bronze/"]
    SQL["SQL Server"]
    BQ["BigQuery"]
    FS["Firestore"]
    LOC -->|"fast_executemany / bcp"| SQL
    LOC -->|"load_table_from_file<br/>bq CLI"| BQ
    LOC -->|"batch.set / BulkWriter"| FS
    GCS -->|"load_table_from_uri"| BQ
    GCS -->|"download_as_bytes +<br/>fast_executemany"| SQL
    SQL -->|"load_table_from_dataframe<br/>(GCS staging)"| BQ
    BQ -->|"to_dataframe +<br/>fast_executemany"| SQL
    SQL -->|"batch.set / BulkWriter"| FS
    SQL -->|"csv.writer"| LOC
    BQ -->|"extract_table"| GCS

Python example implementing the workflow described above.

# Suppress tqdm progress bars globally (pandas_gbq uses tqdm internally)
import os
os.environ['TQDM_DISABLE'] = '1'
import warnings
from sqlalchemy import exc
 
# Suppress SQLAlchemy unrecognized version warnings to keep console output clean
warnings.filterwarnings('ignore', category=exc.SAWarning)
# All imports for data ingestion benchmarking across GCP services
 
# Standard library
import json
import os
import shutil
import subprocess
import tempfile
import time
from datetime import datetime, timezone
from io import BytesIO, StringIO
from pathlib import Path
from urllib.parse import quote_plus
 
# Data & serialisation
import pandas as pd
import pandas_gbq
import pyarrow as pa
import pyarrow.parquet as pq
 
# SQL Server
import pymssql
import pyodbc
from sqlalchemy import create_engine
 
# Google Cloud
from dotenv import load_dotenv
from google.cloud import bigquery, firestore, storage
from google.cloud import bigquery_connection_v1
 
# Visualisation
import plotly.graph_objects as go
import plotly.express as px
from IPython.display import display
 
# Render DataFrames as HTML
html_formatter = get_ipython().display_formatter.formatters['text/html'] # type: ignore
html_formatter.for_type(pd.DataFrame, lambda df: df.to_html())
_ = html_formatter.for_type(pd.Series, lambda s: s.to_frame().to_html())

Python example implementing the workflow described above.

# Load .env and define project constants
load_dotenv(override=True)
 
# ── Project constants ──
PROJECT_ID    = 'seclab-dev-ap-26'
REGION        = 'europe-west1'
BUCKET_NAME   = f'{PROJECT_ID}-data'
BQ_DATASET    = 'index_data'
FIRESTORE_DB  = 'seclab-scores'
SQL_IP        = os.environ['GCP_SQL_IP']
SQL_PASSWORD  = os.environ['GCP_SQL_PASSWORD']
SA_KEY_PATH   = os.environ.get('GCP_SA_KEY_PATH', './gcp-sa-key.json')
DATA_DIR      = Path(r'C:\Users\aperi\DEV\LANG\data')
CHUNK_SIZE    = 10_000
 
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = SA_KEY_PATH
 
# ── Benchmark table/collection names ──
SQL_BENCH_TABLE = 'dbo.ohlcv_bench'
BQ_BENCH_TABLE  = f'{PROJECT_ID}.{BQ_DATASET}.ohlcv_bench'
FS_COLLECTION   = 'ohlcv_bench'
 
# ── GCP clients (authenticated via service account key) ──
bq_client  = bigquery.Client(project=PROJECT_ID)
gcs_client = storage.Client(project=PROJECT_ID)
bucket     = gcs_client.bucket(BUCKET_NAME)
fs_client  = firestore.Client(project=PROJECT_ID, database=FIRESTORE_DB)
 
# ── SQL Server connections ──
#
# pymssql: lightweight FreeTDS-based driver, good for quick queries and scripting.
# pyodbc:  ODBC Driver 18 with TLS encryption — production-grade, supports fast_executemany.
 
def sql_pymssql():
    return pymssql.connect(server=SQL_IP, user='sqlserver', password=SQL_PASSWORD,
                           port='1433', database='stoxx', login_timeout=15)
 
ODBC_CONN = (
    f'DRIVER={{ODBC Driver 18 for SQL Server}};'
    f'SERVER={SQL_IP},1433;DATABASE=stoxx;'
    f'UID=sqlserver;PWD={SQL_PASSWORD};'
    f'Encrypt=yes;TrustServerCertificate=yes;Connection Timeout=15;'
)
def sql_pyodbc():
    return pyodbc.connect(ODBC_CONN)
 
# SQLAlchemy engines (for pandas.to_sql / pandas.read_sql)
sql_engine_odbc   = create_engine(f'mssql+pyodbc:///?odbc_connect={quote_plus(ODBC_CONN)}')
sql_engine_pymssql = create_engine(
    f'mssql+pymssql://sqlserver:{quote_plus(SQL_PASSWORD)}@{SQL_IP}/stoxx')
 
# ── CLI tool paths ──
BCP    = shutil.which('bcp') or 'bcp'
BQ_CLI = shutil.which('bq.cmd') or shutil.which('bq') or 'bq'
GCLOUD = shutil.which('gcloud.cmd') or shutil.which('gcloud') or 'gcloud'
 
# ── Helpers ──
def _sql_truncate(table):
    """Truncate a SQL Server table before benchmark run."""
    with sql_pymssql() as conn:
        conn.cursor().execute(f'TRUNCATE TABLE {table}')
        conn.commit()
 
def _fs_delete_collection(collection, batch_size=500):
    """Delete all documents in a Firestore collection in batches to avoid query timeout."""
    deleted = 0
    while True:
        docs = list(fs_client.collection(collection).limit(batch_size).select([]).stream())
        if not docs:
            break
        batch = fs_client.batch()
        for doc in docs:
            batch.delete(doc.reference)
        batch.commit()
        deleted += len(docs)
    return deleted
 
# ── Test all connections ──
with sql_pymssql() as conn:
    cursor = conn.cursor()
    cursor.execute('SELECT @@VERSION')
    print(f'  SQL Server: {str(cursor.fetchone()[0])[:60]}...')
print(f'  BigQuery:   {BQ_DATASET} ({PROJECT_ID})')
print(f'  Firestore:  {FIRESTORE_DB} ({PROJECT_ID})')
print(f'  GCS:        gs://{BUCKET_NAME}')
  SQL Server: Microsoft SQL Server 2022 (RTM-CU23) (KB5078297) - 16.0.4236...
  index_data (seclab-dev-ap-26)
  seclab-scores (seclab-dev-ap-26)
  gs://seclab-dev-ap-26-data

Python Data Ingestion - SQL Server, BigQuery, Firestore Setup

Helper functions and benchmark infrastructure shared across all ingestion sections. Run these cells once before executing any benchmark.

Helpers and test data

Formatting utilities, tier definitions, and the benchmark recorder used by every ingestion section.

Formatting helpers

Human-readable formatters for row counts, elapsed time, byte sizes, and throughput rates. Used in every benchmark output row and summary table.

Python example implementing the workflow described above.

def fmt_rows(n):
    if n < 1000: return str(n)
    if n < 1_000_000: return f'{n/1000:.1f}K'
    return f'{n/1_000_000:.1f}M'
 
def fmt_time(ms):
    if ms < 1000: return f'{ms:.0f}ms'
    if ms < 60_000: return f'{ms/1000:.1f}s'
    m, s = divmod(ms / 1000, 60)
    if s == 0: return f'{int(m)}min'
    return f'{int(m)}m{s:.0f}s'
 
def fmt_size(b):
    if b < 1024: return f'{b} B'
    if b < 1024 * 1024: return f'{b/1024:.0f} KB'
    if b < 1024 ** 3: return f'{b/(1024*1024):.2f} MB'
    return f'{b/(1024**3):.2f} GB'
 
def fmt_rate(rows, ms):
    if ms <= 0: return '-'
    rate = rows / (ms / 1000)
    if rate < 1000: return f'{rate:.0f} rows/s'
    if rate < 1_000_000: return f'{rate/1000:.1f}K rows/s'
    return f'{rate/1_000_000:.1f}M rows/s'

Define file tiers for ingestion benchmarks

Three file tiers (small 2.5K, medium 75K, large 750K rows) with paths for CSV, JSON, and Parquet formats. The unified OHLCV schema — generated from combined eurostoxx50, stoxxusa50, and oil20 data — is consistent across all tiers and all target systems. GCS staging paths mirror the local tier names under the bronze/ prefix.

Python example implementing the workflow described above.

OHLCV_COLS = ['id', 'symbol', 'date', 'open', 'high', 'low', 'close',
              'adj_close', 'volume', 'dividends', 'stock_splits', 'is_filled']
 
tiers = {
    'small':  {'csv': DATA_DIR / 'ingest_small.csv',  'json': DATA_DIR / 'ingest_small.json',  'parquet': DATA_DIR / 'ingest_small.parquet'},
    'medium': {'csv': DATA_DIR / 'ingest_medium.csv', 'json': DATA_DIR / 'ingest_medium.json', 'parquet': DATA_DIR / 'ingest_medium.parquet'},
    'large':  {'csv': DATA_DIR / 'ingest_large.csv',  'json': DATA_DIR / 'ingest_large.json',  'parquet': DATA_DIR / 'ingest_large.parquet'},
}
 
# Count rows dynamically (header excluded for CSV)
for tier, info in tiers.items():
    with open(info['csv']) as f:
        info['rows'] = sum(1 for _ in f) - 1
 
gcs_paths = {tier: {fmt: f'bronze/{fmt}/ingest_{tier}.{fmt}'
              for fmt in ['csv', 'json', 'parquet']} for tier in tiers}
 
 
tier_summary = pd.DataFrame([
    {
        'tier': tier,
        'rows': f'{info["rows"]:,}',
        'CSV': fmt_size(info['csv'].stat().st_size),
        'JSON': fmt_size(info['json'].stat().st_size),
        'Parquet': fmt_size(info['parquet'].stat().st_size),
    }
    for tier, info in tiers.items()
]).set_index('tier')
display(tier_summary)
rowsCSVJSONParquet
tier
small2,500190 KB468 KB130 KB
medium75,0005.66 MB13.81 MB2.96 MB
large750,00057.31 MB138.85 MB18.89 MB

Ingestion benchmark helper

Times a single ingestion function and upserts the result into an in-memory list and a persistent JSON file, keyed by (method, tier). Re-running a benchmark for the same key overwrites the previous record, keeping the results file stable across partial re-runs.

Python example implementing the workflow described above.

INGEST_RESULTS_FILE = DATA_DIR / 'ingestion_results.json'
 
def _load_results() -> list:
    if INGEST_RESULTS_FILE.exists():
        return json.loads(INGEST_RESULTS_FILE.read_text())
    return []
 
def _save_results(results: list) -> None:
    INGEST_RESULTS_FILE.write_text(json.dumps(results, indent=2))
 
ingest_results = _load_results()
 
def bench_ingest(method_name, ingest_fn, tier_name):
    t0 = time.perf_counter()
    row_count = ingest_fn()
    elapsed_ms = (time.perf_counter() - t0) * 1000
    record = {
        'method': method_name, 'tier': tier_name,
        'rows': row_count, 'rows_fmt': fmt_rows(row_count),
        'elapsed_ms': round(elapsed_ms, 1), 'elapsed': fmt_time(elapsed_ms),
        'rate': fmt_rate(row_count, elapsed_ms),
        'rate_raw': round(row_count / (elapsed_ms / 1000), 1) if elapsed_ms > 0 else 0,
    }
    # Upsert in persistent store
    all_r = _load_results()
    all_r = [r for r in all_r if (r['method'], r['tier']) != (method_name, tier_name)]
    all_r.append(record)
    _save_results(all_r)
    # Upsert in memory
    global ingest_results
    ingest_results = [r for r in ingest_results if (r['method'], r['tier']) != (method_name, tier_name)]
    ingest_results.append(record)
    return record
 
print(f'  Loaded {len(ingest_results)} existing results from {INGEST_RESULTS_FILE.name}')
  Loaded 67 existing results from ingestion_results.json

Schema Setup

Create unified ohlcv_bench staging table in SQL Server and BigQuery. Same OHLCV schema everywhere. Firestore is schemaless — no setup needed. The SQL Server DDL below follows the same bronze-layer-loading patterns used in the medallion architecture, while the BigQuery schema aligns with the format decisions documented in data-loading-and-export.

Staging tables

DDL cells that create the ohlcv_bench table in SQL Server and BigQuery before benchmarks run.

pymssql — create staging table in SQL Server

Creates dbo.ohlcv_bench if it does not already exist. All columns use NVARCHAR(50) to match the bronze-layer loading pattern used in the medallion architecture — type coercion is deferred to the transformation layer.

Python example implementing the workflow described above.

with sql_pymssql() as conn:
    cursor = conn.cursor()
    cursor.execute('''
        IF OBJECT_ID('dbo.ohlcv_bench', 'U') IS NULL
        CREATE TABLE dbo.ohlcv_bench (
            id         nvarchar(50),
            symbol     nvarchar(50),
            date       nvarchar(50),
            [open]     nvarchar(50),
            high       nvarchar(50),
            low        nvarchar(50),
            [close]    nvarchar(50),
            adj_close  nvarchar(50),
            volume     nvarchar(50),
            dividends  nvarchar(50),
            stock_splits nvarchar(50),
            is_filled  nvarchar(50)
        )
    ''')
    conn.commit()

google-cloud-bigquery — create staging table in BigQuery

Creates ohlcv_bench in the index_data dataset with a fully typed schema. create_table(..., exists_ok=True) is idempotent — safe to re-run between benchmark sessions.

Python example implementing the workflow described above.

BQ_BENCH_TABLE = f'{PROJECT_ID}.{BQ_DATASET}.ohlcv_bench'
 
bq_schema = [
    bigquery.SchemaField('id', 'INTEGER'),
    bigquery.SchemaField('symbol', 'STRING'),
    bigquery.SchemaField('date', 'DATE'),
    bigquery.SchemaField('open', 'FLOAT'),
    bigquery.SchemaField('high', 'FLOAT'),
    bigquery.SchemaField('low', 'FLOAT'),
    bigquery.SchemaField('close', 'FLOAT'),
    bigquery.SchemaField('adj_close', 'FLOAT'),
    bigquery.SchemaField('volume', 'INTEGER'),
    bigquery.SchemaField('dividends', 'FLOAT'),
    bigquery.SchemaField('stock_splits', 'FLOAT'),
    bigquery.SchemaField('is_filled', 'BOOLEAN'),
]
 
table = bigquery.Table(BQ_BENCH_TABLE, schema=bq_schema)
table = bq_client.create_table(table, exists_ok=True)

Local → SQL Server Ingestion

Benchmarks local-file-to-SQL-Server ingestion using three strategies: row-by-row executemany (baseline, small tier only), fast_executemany via pyodbc (medium and large tiers), and the bcp CLI utility. Covers CSV, JSON, and Parquet source formats.

Ingestion methods

Each method covers a different driver or protocol path; run all three tiers per method to capture the performance curve.

Ingest CSV into SQL Server from local using pymssql executemany over TDS

Parameterised INSERT, one row per network round-trip. Simple but slow — included as baseline for the small tier only.

pymssql.executemany() — simplest ingestion, one round-trip per row. Parameterised queries prevent injection. Works with any SQL Server. Extremely slow for >5K rows — use fast_executemany or bcp instead.

Python example implementing the workflow described above.

def pymssql_executemany():
    _sql_truncate('ohlcv_bench')
    df = pd.read_csv(tiers['small']['csv'], dtype=str, keep_default_na=False)
    cols = ', '.join(f'[{c}]' for c in df.columns)
    placeholders = ', '.join(['%s'] * len(df.columns))
    sql = f'INSERT INTO dbo.ohlcv_bench ({cols}) VALUES ({placeholders})'
    rows = [tuple(r) for r in df.values]
    with sql_pymssql() as conn:
        cursor = conn.cursor()
        cursor.executemany(sql, rows)
        conn.commit()
    return len(rows)
 
r = bench_ingest('pymssql_executemany', pymssql_executemany, 'small')
print(f'  small    {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  small        2.5K      59.5s      42 rows/s

Ingest CSV into SQL Server from local using pyodbc fast_executemany over ODBC Driver 18 (TLS)

Packs all rows into a single TDS packet. ODBC Driver 18 with TLS encryption. 5-10x faster than plain executemany.

fast_executemany=True packs all rows into a single TDS packet per chunk — 5–10x faster. ODBC Driver 18 with TLS. Chunk at 10K rows to avoid timeouts. For >1M rows, bcp is 2–5x faster.

Always chunk large datasets

Always chunk large datasets — sending all rows in one executemany() causes Communication link failure. Always set fast_executemany=True — without it, falls back to row-by-row.

Safe Pattern: Chunked executemany

Split the DataFrame into chunks of 10 000 rows and call cursor.executemany() per chunk with fast_executemany=True set on the pyodbc connection. Wrap each chunk in a try/except to log the failing range without rolling back the entire load — then re-run only the failed chunks.

Python example implementing the workflow described above.

def pyodbc_fast(tier):
    _sql_truncate('ohlcv_bench')
    df = pd.read_csv(tiers[tier]['csv'], dtype=str, keep_default_na=False)
    cols = ', '.join(f'[{c}]' for c in df.columns)
    placeholders = ', '.join(['?'] * len(df.columns))
    sql = f'INSERT INTO dbo.ohlcv_bench ({cols}) VALUES ({placeholders})'
    rows = [tuple(r) for r in df.values]
    with sql_pyodbc() as conn:
        cursor = conn.cursor()
        cursor.fast_executemany = True
        for start in range(0, len(rows), CHUNK_SIZE):
            cursor.executemany(sql, rows[start:start + CHUNK_SIZE])
        conn.commit()
    return len(rows)
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('pyodbc_fast_executemany', lambda t=tier: pyodbc_fast(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K       1.5s    1.7K rows/s
  medium      75.0K       5.7s   13.0K rows/s
  large      750.0K      54.4s   13.8K rows/s

Ingest CSV into SQL Server from local using bcp (Bulk Copy Program) over TDS

The bcp CLI is the fastest bulk loader for SQL Server. Native TDS bulk-insert protocol — bypasses the SQL parser entirely. Production standard for ETL pipelines.

bcp streams rows via native TDS bulk-insert protocol — bypasses SQL parser, writes directly to table pages. 10–50x faster than row-by-row. Always use -F 2 (skip header), -b 10000 (batch size for recovery).

Don't hardcode passwords in CLI args

Don’t hardcode passwords in CLI args — use -T (trusted) or env vars. Don’t skip -b (batch size) — one failed row rolls back the entire load.

Safe Pattern: Trusted Auth with Batch Recovery

Use -T (Windows Integrated Authentication) or pass credentials via environment variables read at runtime — never interpolated into the shell command string. Always supply -b 10000 so that a bad row aborts only that batch, not the whole load; failed batches are logged and can be re-run in isolation.

Python example implementing the workflow described above.

BCP = shutil.which('bcp') or 'bcp'
 
def bcp_import(tier):
    _sql_truncate('ohlcv_bench')
    result = subprocess.run([
        BCP, 'dbo.ohlcv_bench', 'in', str(tiers[tier]['csv']),
        '-S', f'{SQL_IP},1433', '-U', 'sqlserver', '-P', SQL_PASSWORD,
        '-d', 'stoxx', '-c', '-t', ',', '-F', '2', '-b', '10000', '-u',  # -u = trust server cert (Cloud SQL self-signed)
    ], capture_output=True, text=True)
    if result.returncode != 0:
        print(f'    bcp error: {result.stdout[:200]} {result.stderr[:200]}')
        return 0
    for line in result.stdout.splitlines():
        if 'rows copied' in line.lower():
            return int(line.split()[0])
    return tiers[tier]['rows']
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('bcp_import', lambda t=tier: bcp_import(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K      770ms    3.2K rows/s
  medium      75.0K       2.7s   28.1K rows/s
  large      750.0K      21.0s   35.7K rows/s

Ingest JSON into SQL Server from local using pyodbc fast_executemany over ODBC Driver 18 (TLS)

Reads newline-delimited JSON with pandas, then inserts via fast_executemany. Same TLS-encrypted ODBC path as CSV.

Reads newline-delimited JSON with pandas, converts to strings, inserts via fast_executemany. Handles nested/optional fields. JSON nulls become NaN — must fillna before INSERT.

Python example implementing the workflow described above.

def json_to_sql(tier):
    _sql_truncate('ohlcv_bench')
    df = pd.read_json(tiers[tier]['json'], lines=True, dtype=str)
    df = df.fillna('')
    cols = ', '.join(f'[{c}]' for c in df.columns)
    placeholders = ', '.join(['?'] * len(df.columns))
    sql = f'INSERT INTO dbo.ohlcv_bench ({cols}) VALUES ({placeholders})'
    rows = [tuple(r) for r in df.values]
    with sql_pyodbc() as conn:
        cursor = conn.cursor()
        cursor.fast_executemany = True
        for start in range(0, len(rows), CHUNK_SIZE):
            cursor.executemany(sql, rows[start:start + CHUNK_SIZE])
        conn.commit()
    return len(rows)
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('json_to_sql_fast', lambda t=tier: json_to_sql(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K      722ms    3.5K rows/s
  medium      75.0K       6.2s   12.2K rows/s
  large      750.0K      59.0s   12.7K rows/s

Ingest Parquet into SQL Server from local using pyodbc fast_executemany over ODBC Driver 18 (TLS)

Reads Parquet with pyarrow (fastest local parse), then inserts via fast_executemany. Parquet’s columnar format makes the read near-instant.

Reads Parquet with pyarrow (zero-copy columnar, near-instant), then inserts via fast_executemany. Parquet’s schema is embedded — no column mapping errors. Smallest file size = less disk I/O.

Python example implementing the workflow described above.

def parquet_to_sql(tier):
    _sql_truncate('ohlcv_bench')
    df = pd.read_parquet(tiers[tier]['parquet']).astype(str)
    df = df.replace('nan', '').replace('None', '')
    cols = ', '.join(f'[{c}]' for c in df.columns)
    placeholders = ', '.join(['?'] * len(df.columns))
    sql = f'INSERT INTO dbo.ohlcv_bench ({cols}) VALUES ({placeholders})'
    rows = [tuple(r) for r in df.values]
    with sql_pyodbc() as conn:
        cursor = conn.cursor()
        cursor.fast_executemany = True
        for start in range(0, len(rows), CHUNK_SIZE):
            cursor.executemany(sql, rows[start:start + CHUNK_SIZE])
        conn.commit()
    return len(rows)
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('parquet_to_sql_fast', lambda t=tier: parquet_to_sql(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K      837ms    3.0K rows/s
  medium      75.0K       5.8s   12.9K rows/s
  large      750.0K      55.2s   13.6K rows/s

Ingest CSV into SQL Server using BULK INSERT (T-SQL) and SSMS Import Wizard (reference)

BULK INSERT: Native T-SQL command. Requires the file to be accessible from the server filesystem — not supported on Cloud SQL (server can’t read client-side files). Use bcp instead.

SSMS / Azure Data Studio Import Wizard: Graphical import via right-click → Tasks → Import Flat File. Best for ad-hoc one-off imports under 100K rows. Not benchmarkable from a notebook.

Local → BigQuery Ingestion

Load data from local files into BigQuery. Three formats (CSV, JSON, Parquet), plus bq CLI and Storage Write API.

Ingestion methods

Covers the Python client library, bq CLI, pandas-gbq Storage Write API, and read-only external tables.

Ingest CSV into BigQuery from local using google-cloud-bigquery load_table_from_file over HTTPS

Server parses CSV rows. skip_leading_rows=1 for header. WRITE_TRUNCATE clears before load.

Uploads CSV to BigQuery’s load job API over HTTPS. Server-side parsing, atomic (fully succeeds or fails). Always set skip_leading_rows=1. Use explicit schema in production (not autodetect). For large files, use Parquet (5x compression) or GCS staging.

autodetect=True is unsafe in production

BigQuery infers types from the first few rows — a column with nulls early in the file may be typed as STRING instead of FLOAT, causing silent data loss downstream.

Provide an explicit schema in production

Pass schema=bq_schema to LoadJobConfig and set autodetect=False. The schema defined in the Setup section already captures correct types for all OHLCV columns.

Python example implementing the workflow described above.

def bq_load_csv(tier):
    path = DATA_DIR / f'ingest_{tier}.csv'
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.CSV,
        skip_leading_rows=1,
        autodetect=True,
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    )
    with open(path, 'rb') as f:
        job = bq_client.load_table_from_file(f, BQ_BENCH_TABLE, job_config=job_config)
    job.result()
    return job.output_rows
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('bq_load_csv', lambda t=tier: bq_load_csv(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K       3.1s     815 rows/s
  medium      75.0K       6.8s   11.1K rows/s
  large      750.0K      22.7s   33.1K rows/s

Ingest JSON into BigQuery from local using google-cloud-bigquery load_table_from_file over HTTPS

Server parses newline-delimited JSON. Auto-detects schema from keys.

NDJSON (one JSON object per line) — handles nested/repeated fields natively. Schema auto-detected from keys. Use for API dumps or nested data. For flat tabular data, CSV/Parquet is 3–5x smaller.

Python example implementing the workflow described above.

def bq_load_json(tier):
    path = DATA_DIR / f'ingest_{tier}.json'
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON,
        autodetect=True,
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    )
    with open(path, 'rb') as f:
        job = bq_client.load_table_from_file(f, BQ_BENCH_TABLE, job_config=job_config)
    job.result()
    return job.output_rows
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('bq_load_json', lambda t=tier: bq_load_json(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K       4.5s     554 rows/s
  medium      75.0K       7.3s   10.3K rows/s
  large      750.0K      27.7s   27.1K rows/s

Ingest Parquet into BigQuery from local using google-cloud-bigquery load_table_from_file over HTTPS

Fastest format — columnar, compressed, schema embedded. No parsing overhead.

Parquet schema maps directly to BQ schema — no parsing, no autodetect needed. Snappy compression = smallest upload. Type-safe end-to-end. BigQuery-recommended format for production pipelines.

Python example implementing the workflow described above.

def bq_load_parquet(tier):
    path = DATA_DIR / f'ingest_{tier}.parquet'
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.PARQUET,
 
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    )
    with open(path, 'rb') as f:
        job = bq_client.load_table_from_file(f, BQ_BENCH_TABLE, job_config=job_config)
    job.result()
    return job.output_rows
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('bq_load_parquet', lambda t=tier: bq_load_parquet(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K       2.8s     884 rows/s
  medium      75.0K       3.6s   20.8K rows/s
  large      750.0K       8.3s   90.2K rows/s

Ingest CSV into BigQuery from local using bq CLI bq load over HTTPS

Command-line tool — same load job API but no Python code needed.

bq load — same load job API as Python but no code needed. Just the gcloud SDK. Good for shell scripts, CI/CD, and one-off loads.

Python example implementing the workflow described above.

BQ_CMD = shutil.which('bq.cmd') or shutil.which('bq') or 'bq'
 
def bq_cli_csv(tier):
    table_id = f'{BQ_DATASET}.ohlcv_bench'
    result = subprocess.run([
        BQ_CMD, 'load', '--source_format=CSV', '--skip_leading_rows=1',
        '--autodetect', '--replace', f'--project_id={PROJECT_ID}',
        table_id, str(tiers[tier]['csv']),
    ], capture_output=True, text=True)
    if result.returncode != 0:
        print(f'    bq error: {result.stderr[:200]}')
        return 0
    return tiers[tier]['rows']
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('bq_cli_csv', lambda t=tier: bq_cli_csv(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K       9.0s     278 rows/s
  medium      75.0K       7.9s    9.5K rows/s
  large      750.0K      19.7s   38.0K rows/s

Ingest data into BigQuery using pandas-gbq Storage Write API over gRPC

Highest throughput for streaming ingestion. pandas_gbq.to_gbq() uses the Storage Write API when available.

pandas_gbq.to_gbq() uses the Storage Write API (gRPC) — writes directly to BQ storage with row-level acknowledgment and exactly-once semantics. Highest throughput for streaming. For batch loads, load jobs are simpler.

Python example implementing the workflow described above.

def bq_storage_write(tier):
    df = pd.read_csv(tiers[tier]['csv'])
    pandas_gbq.to_gbq(df, f'{BQ_DATASET}.ohlcv_bench', project_id=PROJECT_ID, if_exists='replace')
    return len(df)
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('bq_storage_write', lambda t=tier: bq_storage_write(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K       4.7s     527 rows/s
  medium      75.0K       4.0s   18.8K rows/s
  large      750.0K      11.8s   63.6K rows/s

Query data from GCS without loading using BigQuery External Tables over internal network

Query CSV/JSON/Parquet in GCS directly via SQL. Zero ingestion time — slower queries but no storage cost.

External table points to a GCS file — queries read directly at query time. Zero ingestion, zero storage cost. 10–100x slower than native tables. Use for ad-hoc exploration; not for production dashboards.

Use external tables for one-off exploration, not production dashboards

External tables re-scan the GCS file on every query. For recurring workloads or dashboards, load the data into a native BigQuery table first — queries will be 10–100x faster and cheaper.

Python example implementing the workflow described above.

def bq_external_table(tier):
    ext_table = f'{PROJECT_ID}.{BQ_DATASET}.ohlcv_bench_ext'
    uri = f'gs://{BUCKET_NAME}/{gcs_paths[tier]["parquet"]}'
    ext_config = bigquery.ExternalConfig('PARQUET')
    ext_config.source_uris = [uri]
    table = bigquery.Table(ext_table)
    table.external_data_configuration = ext_config
    table = bq_client.create_table(table, exists_ok=True)
    result = bq_client.query(f'SELECT COUNT(*) as cnt FROM `{ext_table}`').result()
    row_count = list(result)[0].cnt
    bq_client.delete_table(ext_table, not_found_ok=True)
    return row_count
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('bq_external_table', lambda t=tier: bq_external_table(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K       1.1s    2.2K rows/s
  medium      75.0K       1.2s   64.1K rows/s
  large      750.0K       9.7s   77.6K rows/s

Ingest data into BigQuery using Google Cloud Console Web UI (reference)

Console → BigQuery → Dataset → Create Table → Upload (up to 10 MB) or Google Cloud Storage. Not benchmarkable from a notebook.

Local → Firestore Ingestion

Write OHLCV data into Firestore. Each row becomes a document in the ohlcv_bench collection.

Ingestion methods

Compares manual batch.set() (500-doc limit) against BulkWriter (parallel, auto-throttled).

Ingest CSV into Firestore from local using google-cloud-firestore batch.set over gRPC

500-doc batches (Firestore limit). Each batch is a single gRPC call.

500-doc batches (Firestore’s limit). Each batch.commit() is a single atomic gRPC call. Simple API. For >50K docs, BulkWriter is 2–5x faster (parallel batches).

Don't exceed 500 docs per

Don’t exceed 500 docs per batch (rejected). Don’t forget to commit the final partial batch.

Safe Pattern: Flush on Limit

Track a counter alongside the batch; when it reaches 500, call batch.commit() and immediately open a new fs_client.batch(). After the loop, check whether the counter is non-zero and commit the final partial batch — this guarantees no rows are silently dropped even when the total is not an exact multiple of 500.

Python example implementing the workflow described above.

def fs_batch_write(tier):
    df = pd.read_csv(tiers[tier]['csv'], dtype=str, keep_default_na=False)
    batch = fs_client.batch()
    count = 0
    for idx, row in df.iterrows():
        doc_ref = fs_client.collection(FS_COLLECTION).document(str(idx))
        batch.set(doc_ref, row.to_dict())
        count += 1
        if count % 500 == 0:
            batch.commit()
            batch = fs_client.batch()
    if count % 500 != 0:
        batch.commit()
    return count
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('fs_batch', lambda t=tier: fs_batch_write(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K       6.7s     372 rows/s
  medium      75.0K      3m21s     372 rows/s
  large      750.0K     31m13s     400 rows/s

Ingest CSV into Firestore from local using google-cloud-firestore BulkWriter over gRPC

BulkWriter manages batching, retries, and throttling automatically. Parallel writes — the recommended method for bulk ingestion.

BulkWriter manages batching, retries, and rate limiting automatically. 2–5x faster than manual batch.set(). Use for 10K–500K document migrations/backfills. set() overwrites existing documents by ID — no prior delete is needed, which avoids a costly collection scan.

Always call bw.close() before the function returns

BulkWriter buffers writes internally. If the process exits or an exception is raised before close(), buffered documents are silently lost with no error raised.

Wrap in try/finally to guarantee flush 

bw = fs_client.bulk_writer()
try:
    for idx, row in df.iterrows():
        bw.set(collection.document(str(idx)), row.to_dict())
finally:
    bw.close()

Python example implementing the workflow described above.

def fs_bulk_write(tier):
    df = pd.read_csv(tiers[tier]['csv'], dtype=str, keep_default_na=False)
    bw = fs_client.bulk_writer()
    count = 0
    for idx, row in df.iterrows():
        doc_ref = fs_client.collection(FS_COLLECTION).document(str(idx))
        bw.set(doc_ref, row.to_dict())
        count += 1
    bw.close()
    return count
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('fs_bulkwriter', lambda t=tier: fs_bulk_write(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K       4.1s     604 rows/s
  medium      75.0K      2m31s     497 rows/s
  large      750.0K     25m11s     496 rows/s

Import data into Firestore from GCS using gcloud firestore import and Console UI (reference)

gcloud firestore export/import: Managed backup/restore from GCS. Server-side, fastest for large restores.

Firebase/GCP Console: Manual document creation or import from managed exports.

Both require managed export format (not raw CSV/JSON).

GCS → BigQuery Ingestion

Server-side operation — no data passes through the local machine.

Ingestion methods

Three source formats (CSV, JSON, Parquet) loaded via load_table_from_uri, all flowing within Google’s network.

Ingest CSV into BigQuery from GCS using google-cloud-bigquery load_table_from_uri over internal network

Server-side CSV parse. Data flows GCS → BigQuery within Google’s network.

load_table_from_uri triggers a server-side load — data flows GCS → BQ within Google’s network, no local bandwidth. Supports wildcards (gs://bucket/path/*.csv). Standard data lake pattern.

Python example implementing the workflow described above.

def bq_gcs_csv(tier):
    uri = f'gs://{BUCKET_NAME}/{gcs_paths[tier]["csv"]}'
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.CSV,
        skip_leading_rows=1,
        autodetect=True,
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    )
    job = bq_client.load_table_from_uri(uri, BQ_BENCH_TABLE, job_config=job_config)
    job.result()
    return job.output_rows
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('bq_gcs_csv', lambda t=tier: bq_gcs_csv(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K       3.2s     775 rows/s
  medium      75.0K       4.8s   15.7K rows/s
  large      750.0K      13.0s   57.9K rows/s

Ingest JSON into BigQuery from GCS using google-cloud-bigquery load_table_from_uri over internal network

Server-side JSON parse. Same internal network path as CSV — data flows GCS → BigQuery within Google’s network.

Python example implementing the workflow described above.

def bq_gcs_json(tier):
    uri = f'gs://{BUCKET_NAME}/{gcs_paths[tier]["json"]}'
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON,
        autodetect=True,
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    )
    job = bq_client.load_table_from_uri(uri, BQ_BENCH_TABLE, job_config=job_config)
    job.result()
    return job.output_rows
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('bq_gcs_json', lambda t=tier: bq_gcs_json(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K       4.1s     614 rows/s
  medium      75.0K       6.6s   11.3K rows/s
  large      750.0K      17.6s   42.6K rows/s

Ingest Parquet into BigQuery from GCS using google-cloud-bigquery load_table_from_uri over internal network

Fastest GCS source format — columnar, compressed, schema embedded, no server-side parsing overhead.

Python example implementing the workflow described above.

def bq_gcs_parquet(tier):
    uri = f'gs://{BUCKET_NAME}/{gcs_paths[tier]["parquet"]}'
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.PARQUET,
 
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    )
    job = bq_client.load_table_from_uri(uri, BQ_BENCH_TABLE, job_config=job_config)
    job.result()
    return job.output_rows
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('bq_gcs_parquet', lambda t=tier: bq_gcs_parquet(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K       2.5s    1.0K rows/s
  medium      75.0K       2.9s   26.1K rows/s
  large      750.0K       7.6s   98.9K rows/s

GCS → SQL Server Ingestion

Two-hop: download from GCS to memory, then insert into SQL Server.

Ingestion methods

Single method: download CSV from GCS into memory via BytesIO, then insert with fast_executemany.

Ingest CSV into SQL Server from GCS using google-cloud-storage download + pyodbc fast_executemany over HTTPS + TLS

Download CSV → pandas → fast_executemany. Combined pipeline.

Two-hop: download CSV from GCS into memory (BytesIO), parse with pandas, insert via fast_executemany in 10K-row chunks. No direct GCS→SQL path exists. For GB-scale, use VM-hosted SQL Server or Dataflow instead.

Python example implementing the workflow described above.

def gcs_csv_to_sql(tier):
    _sql_truncate('ohlcv_bench')
    blob = bucket.blob(gcs_paths[tier]['csv'])
    csv_bytes = blob.download_as_bytes()
    df = pd.read_csv(BytesIO(csv_bytes), dtype=str, keep_default_na=False)
    cols = ', '.join(f'[{c}]' for c in df.columns)
    placeholders = ', '.join(['?'] * len(df.columns))
    sql = f'INSERT INTO dbo.ohlcv_bench ({cols}) VALUES ({placeholders})'
    rows = [tuple(r) for r in df.values]
    with sql_pyodbc() as conn:
        cursor = conn.cursor()
        cursor.fast_executemany = True
        for start in range(0, len(rows), CHUNK_SIZE):
            cursor.executemany(sql, rows[start:start + CHUNK_SIZE])
        conn.commit()
    return len(rows)
 
print(f'  {"tier":<8s} {"rows":>8s} {"time":>10s} {"rate":>14s}')
for tier in tiers:
    r = bench_ingest('gcs_csv_to_sql', lambda t=tier: gcs_csv_to_sql(t), tier)
    print(f'  {tier:<8s} {r["rows_fmt"]:>8s} {r["elapsed"]:>10s} {r["rate"]:>14s}')
  tier         rows       time           rate
  small        2.5K       2.2s    1.1K rows/s
  medium      75.0K       6.5s   11.5K rows/s
  large      750.0K      56.2s   13.3K rows/s

Cross-Service Transfers

Move data between SQL Server, BigQuery, and Firestore.

Transfer methods

Five bidirectional paths covering SQL↔BigQuery, SQL→Firestore, BQ→Firestore, and the GCS-staged SQL→BigQuery route.

Transfer data from SQL Server to BigQuery using pymssql query + load_table_from_dataframe over TDS + HTTPS

Query SQL Server → DataFrame → BigQuery. Two-hop via local memory. Self-populates SQL Server first (Step 1: load CSV), then queries and transfers to BigQuery (Step 2).

Python example implementing the workflow described above.

def sql_to_bq(tier):
    # Step 1: populate SQL Server with this tier's data
    _sql_truncate('ohlcv_bench')
    df = pd.read_csv(tiers[tier]['csv'], dtype=str, keep_default_na=False)
    df.to_sql('ohlcv_bench', con=sql_engine_pymssql, schema='dbo', if_exists='append', index=False)
 
    # Step 2: read from SQL Server, load into BigQuery
    query = f'SELECT TOP {tiers[tier]["rows"]} * FROM dbo.ohlcv_bench'
    df2 = pd.read_sql(query, con=sql_engine_pymssql)
    job_config = bigquery.LoadJobConfig(
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
        autodetect=True
    )
    job = bq_client.load_table_from_dataframe(df2, BQ_BENCH_TABLE, job_config=job_config)
    job.result()
    return job.output_rows
 
print(f"  {'tier':<8s} {'rows':>8s} {'time':>10s} {'rate':>14s}")
for tier in tiers:
    r = bench_ingest('sql_to_bq', lambda t=tier: sql_to_bq(t), tier)
    print(f"  {tier:<8s} {r['rows_fmt']:>8s} {r['elapsed']:>10s} {r['rate']:>14s}")
  tier         rows       time           rate
  small        2.5K       7.4s     340 rows/s
  medium      75.0K      51.8s    1.4K rows/s
  large      750.0K      8m19s    1.5K rows/s

Transfer data from BigQuery to SQL Server using google-cloud-bigquery query + pyodbc fast_executemany over HTTPS + TLS

Query BigQuery → DataFrame → SQL Server. Self-populates BigQuery first (Step 1), then queries and inserts into SQL Server via fast_executemany (Step 2).

Python example implementing the workflow described above.

def bq_to_sql(tier):
    # Step 1: populate BigQuery with this tier's data
    df = pd.read_csv(tiers[tier]['csv'], dtype=str, keep_default_na=False)
    job_config = bigquery.LoadJobConfig(
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
        autodetect=True
    )
    job = bq_client.load_table_from_dataframe(df, BQ_BENCH_TABLE, job_config=job_config)
    job.result()
 
    # Step 2: query BigQuery, insert into SQL Server
    _sql_truncate('ohlcv_bench')
    results = bq_client.query(f'SELECT * FROM `{PROJECT_ID}.{BQ_DATASET}.ohlcv_bench`')
    rows = [dict(row) for row in results]
    df2 = pd.DataFrame(rows).astype(str)
    df2.to_sql('ohlcv_bench', con=sql_engine_pymssql, schema='dbo', if_exists='append', index=False)
    return len(rows)
 
print(f"  {'tier':<8s} {'rows':>8s} {'time':>10s} {'rate':>14s}")
for tier in tiers:
    r = bench_ingest('bq_to_sql', lambda t=tier: bq_to_sql(t), tier)
    print(f"  {tier:<8s} {r['rows_fmt']:>8s} {r['elapsed']:>10s} {r['rate']:>14s}")
  tier         rows       time           rate
  small        2.5K       5.5s     451 rows/s
  medium      75.0K      57.4s    1.3K rows/s
  large      750.0K      8m34s    1.5K rows/s

Transfer data from BigQuery to Firestore using google-cloud-bigquery query + BulkWriter over HTTPS + gRPC

Query BigQuery → iterate results → Firestore BulkWriter. Self-populates BigQuery first (Step 1), then streams rows into Firestore via BulkWriter (Step 2). For real-time serving of scored data.

Python example implementing the workflow described above.

def bq_to_fs(tier):
    # Step 1: populate BigQuery with this tier's data
    df = pd.read_csv(tiers[tier]['csv'], dtype=str, keep_default_na=False)
    job_config = bigquery.LoadJobConfig(
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
        autodetect=True
    )
    job = bq_client.load_table_from_dataframe(df, BQ_BENCH_TABLE, job_config=job_config)
    job.result()
 
    # Step 2: query BigQuery, write to Firestore
    results = bq_client.query(f'SELECT * FROM `{PROJECT_ID}.{BQ_DATASET}.ohlcv_bench`')
    bw = fs_client.bulk_writer()
    count = 0
    for row in results:
        doc_ref = fs_client.collection(FS_COLLECTION).document(str(count))
        bw.set(doc_ref, {k: str(v) for k, v in dict(row).items()})
        count += 1
    bw.close()
    return count
 
print(f"  {'tier':<8s} {'rows':>8s} {'time':>10s} {'rate':>14s}")
for tier in tiers:
    r = bench_ingest('bq_to_fs', lambda t=tier: bq_to_fs(t), tier)
    print(f"  {tier:<8s} {r['rows_fmt']:>8s} {r['elapsed']:>10s} {r['rate']:>14s}")
  tier         rows       time           rate
  small        2.5K       8.9s     282 rows/s
  medium      75.0K      2m41s     466 rows/s
  large      750.0K     25m52s     483 rows/s

Transfer data from SQL Server to Firestore using pymssql query + BulkWriter over TDS + gRPC

Direct SQL Server → Firestore bridge. Self-populates SQL Server first (Step 1), then streams rows into Firestore via BulkWriter (Step 2).

Python example implementing the workflow described above.

def sql_to_fs(tier):
    # Step 1: populate SQL Server with this tier's data
    _sql_truncate('ohlcv_bench')
    df = pd.read_csv(tiers[tier]['csv'], dtype=str, keep_default_na=False)
    df.to_sql('ohlcv_bench', con=sql_engine_pymssql, schema='dbo', if_exists='append', index=False)
 
    # Step 2: query SQL Server, write to Firestore
    with sql_pymssql() as conn:
        cursor = conn.cursor(as_dict=True)
        cursor.execute(f'SELECT TOP {tiers[tier]["rows"]} * FROM dbo.ohlcv_bench')
        bw = fs_client.bulk_writer()
        count = 0
        for row in cursor:
            doc_ref = fs_client.collection(FS_COLLECTION).document(str(count))
            bw.set(doc_ref, {k: str(v) for k, v in row.items()})
            count += 1
        bw.close()
    return count
 
print(f"  {'tier':<8s} {'rows':>8s} {'time':>10s} {'rate':>14s}")
for tier in tiers:
    r = bench_ingest('sql_to_firestore', lambda t=tier: sql_to_fs(t), tier)
    print(f"  {tier:<8s} {r['rows_fmt']:>8s} {r['elapsed']:>10s} {r['rate']:>14s}")
  tier         rows       time           rate
  small        2.5K       6.5s     387 rows/s
  medium      75.0K      3m16s     383 rows/s
  large      750.0K     32m51s     381 rows/s

Transfer data from SQL Server to BigQuery via GCS staging using pandas + GCS + load_table_from_uri

Production pattern: SQL → Parquet → GCS → BigQuery. Avoids local memory bottleneck for large datasets. For this benchmark the function self-populates SQL Server first (Step 1), then exports to GCS and loads into BigQuery (Step 2).

Prefer GCS staging for large SQL → BigQuery migrations

Loading directly from the client via load_table_from_dataframe requires the full dataset to pass through local memory and network. The GCS staging path (SQL → Parquet → GCS → load_table_from_uri) is server-side from GCS onward and scales to hundreds of GB without memory pressure.

Python example implementing the workflow described above.

def sql_to_bq_gcs(tier):
    # Step 1: populate SQL Server with this tier's data
    _sql_truncate('ohlcv_bench')
    df = pd.read_csv(tiers[tier]['csv'], dtype=str, keep_default_na=False)
    df.to_sql('ohlcv_bench', con=sql_engine_pymssql, schema='dbo', if_exists='append', index=False)
 
    # Step 2: query SQL Server, write CSV to GCS, load into BigQuery
    query = f'SELECT TOP {tiers[tier]["rows"]} * FROM dbo.ohlcv_bench'
    df2 = pd.read_sql(query, con=sql_engine_pymssql)
    gcs_path = f'staging/sql_to_bq_{tier}.csv'
    blob = bucket.blob(gcs_path)
    blob.upload_from_string(df2.to_csv(index=False), content_type='text/csv')
    uri = f'gs://{BUCKET_NAME}/{gcs_path}'
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.CSV,
        skip_leading_rows=1,
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
        autodetect=True
    )
    job = bq_client.load_table_from_uri(uri, BQ_BENCH_TABLE, job_config=job_config)
    job.result()
    blob.delete()
    return job.output_rows
 
print(f"  {'tier':<8s} {'rows':>8s} {'time':>10s} {'rate':>14s}")
for tier in tiers:
    r = bench_ingest('sql_to_bq_gcs', lambda t=tier: sql_to_bq_gcs(t), tier)
    print(f"  {tier:<8s} {r['rows_fmt']:>8s} {r['elapsed']:>10s} {r['rate']:>14s}")
  tier         rows       time           rate
  small        2.5K       6.5s     382 rows/s
  medium      75.0K      54.4s    1.4K rows/s
  large      750.0K       8m7s    1.5K rows/s

Export

Export data from SQL Server, BigQuery, and Firestore.

Export methods

Three targets: SQL Server to local CSV, BigQuery to GCS, and Firestore collection to NDJSON.

Export SQL Server to CSV using pandas read_sql + to_csv over TDS

Query into DataFrame, write to local CSV.

Python example implementing the workflow described above.

EXPORT_DIR = DATA_DIR / 'exports'
EXPORT_DIR.mkdir(exist_ok=True)
 
def sql_export(tier):
    query = f'SELECT TOP {tiers[tier]["rows"]} * FROM dbo.ohlcv_bench'
    df = pd.read_sql(query, con=sql_engine_pymssql)
    df.to_csv(EXPORT_DIR / f'export_{tier}.csv', index=False)
    return len(df)
 
print(f"  {'tier':<8s} {'rows':>8s} {'time':>10s} {'rate':>14s}")
for tier in tiers:
    r = bench_ingest('sql_export_csv', lambda t=tier: sql_export(t), tier)
    print(f"  {tier:<8s} {r['rows_fmt']:>8s} {r['elapsed']:>10s} {r['rate']:>14s}")
  tier         rows       time           rate
  small        2.5K      131ms   19.1K rows/s
  medium      75.0K      800ms   93.7K rows/s
  large      750.0K       6.5s  114.7K rows/s

Export BigQuery to GCS using google-cloud-bigquery extract_table over internal network

Server-side export — BigQuery writes directly to GCS.

Python example implementing the workflow described above.

def bq_export(tier):
    row_limit = tiers[tier]['rows']
    # Create a temp table with the correct row count
    temp_table = f'{PROJECT_ID}.{BQ_DATASET}.export_temp_{tier}'
    query = f'CREATE OR REPLACE TABLE `{temp_table}` AS SELECT * FROM `{PROJECT_ID}.{BQ_DATASET}.ohlcv_bench` LIMIT {row_limit}'
    bq_client.query(query).result()
    dest_uri = f'gs://{BUCKET_NAME}/exports/ohlcv_{tier}.csv'
    job_config = bigquery.ExtractJobConfig(destination_format=bigquery.DestinationFormat.CSV)
    job = bq_client.extract_table(temp_table, dest_uri, job_config=job_config)
    job.result()
    row_count = bq_client.get_table(temp_table).num_rows
    bq_client.delete_table(temp_table, not_found_ok=True)
    return row_count
 
print(f"  {'tier':<8s} {'rows':>8s} {'time':>10s} {'rate':>14s}")
for tier in tiers:
    r = bench_ingest('bq_export_gcs', lambda t=tier: bq_export(t), tier)
    print(f"  {tier:<8s} {r['rows_fmt']:>8s} {r['elapsed']:>10s} {r['rate']:>14s}")
  tier         rows       time           rate
  small        2.5K       4.9s     510 rows/s
  medium      75.0K       5.2s   14.3K rows/s
  large      750.0K      15.8s   47.4K rows/s

Export Firestore to JSON using google-cloud-firestore collection.stream over gRPC

Stream documents, write as NDJSON.

Python example implementing the workflow described above.

def fs_export(tier):
    path = EXPORT_DIR / f'ohlcv_firestore_{tier}.json'
    target = tiers[tier]['rows']
    count = 0
    batch_size = 500
    last_doc = None
    with open(path, 'w') as f:
        while count < target:
            query = fs_client.collection(FS_COLLECTION).order_by('__name__').limit(batch_size)
            if last_doc:
                query = query.start_after(last_doc)
            docs = list(query.stream())
            if not docs:
                break
            for doc in docs:
                d = doc.to_dict()
                for k, v in d.items():
                    if hasattr(v, 'isoformat'): d[k] = v.isoformat()
                f.write(json.dumps(d) + chr(10))
                count += 1
                if count >= target:
                    break
            last_doc = docs[-1]
    return count
 
print(f"  {'tier':<8s} {'rows':>8s} {'time':>10s} {'rate':>14s}")
for tier in tiers:
    r = bench_ingest('fs_export_json', lambda t=tier: fs_export(t), tier)
    print(f"  {tier:<8s} {r['rows_fmt']:>8s} {r['elapsed']:>10s} {r['rate']:>14s}")
  tier         rows       time           rate
  small        2.5K      707ms    3.5K rows/s
  medium      75.0K      17.0s    4.4K rows/s
  large      750.0K      2m45s    4.5K rows/s

Summary

Reloads all persisted benchmark results from JSON and renders comparative bar charts for each target system and transfer type. Run this section at any time to visualise results from previous sessions without re-running ingestion.

Results

Benchmark data and Plotly charts grouped by target system (BigQuery, SQL Server, Firestore, cross-service, export).

Results table

Loads raw results from the persistent JSON file, deduplicates to the latest run per (method, tier) combination, and creates a numeric sort key so large-tier rows appear first in charts.

Python example implementing the workflow described above.

all_results = _load_results()
df_results = pd.DataFrame(all_results).drop_duplicates(subset=['method', 'tier'], keep='last')
 
tier_order = {'large': 0, 'medium': 1, 'small': 2}
df_results['tier_rank'] = df_results['tier'].map(tier_order)

BigQuery ingestion benchmark (local and GCS)

This comparison isolates BigQuery ingestion paths so load_table_from_file, load_table_from_uri, and pandas_gbq.to_gbq() can be evaluated on the same throughput axis before you interpret the chart.

Python example implementing the workflow described above.

bq_methods = [
    'bq_load_parquet', 'bq_storage_write', 'bq_gcs_csv', 'bq_external_table',
    'bq_gcs_parquet', 'bq_gcs_json', 'bq_cli_csv', 'bq_load_csv', 'bq_load_json'
]
 
df_bq = df_results[df_results['method'].isin(bq_methods)].copy()
df_bq = df_bq.sort_values(['tier_rank', 'rate_raw'], ascending=[True, False])
 
display(df_bq[['method', 'tier', 'rows_fmt', 'elapsed', 'rate']].reset_index(drop=True))
methodtierrows_fmtelapsedrate
0bq_gcs_parquetlarge750.0K7.6s98.9K rows/s
1bq_load_parquetlarge750.0K8.3s90.2K rows/s
2bq_external_tablelarge750.0K9.7s77.6K rows/s
3bq_storage_writelarge750.0K11.8s63.6K rows/s
4bq_gcs_csvlarge750.0K13.0s57.9K rows/s
5bq_gcs_jsonlarge750.0K17.6s42.6K rows/s
6bq_cli_csvlarge750.0K19.7s38.0K rows/s
7bq_load_csvlarge750.0K22.7s33.1K rows/s
8bq_load_jsonlarge750.0K27.7s27.1K rows/s
9bq_external_tablemedium75.0K1.2s64.1K rows/s
10bq_gcs_parquetmedium75.0K2.9s26.1K rows/s
11bq_load_parquetmedium75.0K3.6s20.8K rows/s
12bq_storage_writemedium75.0K4.0s18.8K rows/s
13bq_gcs_csvmedium75.0K4.8s15.7K rows/s
14bq_gcs_jsonmedium75.0K6.6s11.3K rows/s
15bq_load_csvmedium75.0K6.8s11.1K rows/s
16bq_load_jsonmedium75.0K7.3s10.3K rows/s
17bq_cli_csvmedium75.0K7.9s9.5K rows/s
18bq_external_tablesmall2.5K1.1s2.2K rows/s
19bq_gcs_parquetsmall2.5K2.5s1.0K rows/s
20bq_load_parquetsmall2.5K2.8s884 rows/s
21bq_load_csvsmall2.5K3.1s815 rows/s
22bq_gcs_csvsmall2.5K3.2s775 rows/s
23bq_gcs_jsonsmall2.5K4.1s614 rows/s
24bq_load_jsonsmall2.5K4.5s554 rows/s
25bq_storage_writesmall2.5K4.7s527 rows/s
26bq_cli_csvsmall2.5K9.0s278 rows/s

Python example implementing the workflow described above.

fig_bq = go.Figure()
 
# Iterate through tiers in order to maintain consistent legend grouping
for tier in ['large', 'medium', 'small']:
    df_t = df_bq[df_bq['tier'] == tier]
    if not df_t.empty:
        fig_bq.add_trace(go.Bar(
            x=df_t['method'],
            y=df_t['rate_raw'],
            name=tier.capitalize()
        ))
 
fig_bq.update_layout(
    barmode='group',
    title='BigQuery Ingestion Performance (Log Scale)',
    xaxis_title='Ingestion Method',
    yaxis_title='Speed (Rows / Second)',
    yaxis_type='log',
    template='plotly_dark',
    xaxis_tickangle=-45
)
fig_bq.show()

SQL Server ingestion benchmark (local and GCS)

This view keeps the SQL Server load paths together so bcp, fast_executemany, and file-format-specific variants can be compared without cross-service noise.

Python example implementing the workflow described above.

sql_methods = [
    'bcp_import', 'pyodbc_fast_executemany', 'gcs_csv_to_sql',
    'parquet_to_sql_fast', 'json_to_sql_fast', 'pymssql_executemany'
]
 
df_sql = df_results[df_results['method'].isin(sql_methods)].copy()
df_sql = df_sql.sort_values(['tier_rank', 'rate_raw'], ascending=[True, False])
 
display(df_sql[['method', 'tier', 'rows_fmt', 'elapsed', 'rate']].reset_index(drop=True))
methodtierrows_fmtelapsedrate
0bcp_importlarge750.0K21.0s35.7K rows/s
1pyodbc_fast_executemanylarge750.0K54.4s13.8K rows/s
2parquet_to_sql_fastlarge750.0K55.2s13.6K rows/s
3gcs_csv_to_sqllarge750.0K56.2s13.3K rows/s
4json_to_sql_fastlarge750.0K59.0s12.7K rows/s
5bcp_importmedium75.0K2.7s28.1K rows/s
6pyodbc_fast_executemanymedium75.0K5.7s13.0K rows/s
7parquet_to_sql_fastmedium75.0K5.8s12.9K rows/s
8json_to_sql_fastmedium75.0K6.2s12.2K rows/s
9gcs_csv_to_sqlmedium75.0K6.5s11.5K rows/s
10json_to_sql_fastsmall2.5K722ms3.5K rows/s
11bcp_importsmall2.5K770ms3.2K rows/s
12parquet_to_sql_fastsmall2.5K837ms3.0K rows/s
13pyodbc_fast_executemanysmall2.5K1.5s1.7K rows/s
14gcs_csv_to_sqlsmall2.5K2.2s1.1K rows/s
15pymssql_executemanysmall2.5K59.5s42 rows/s

Python example implementing the workflow described above.

fig_sql = go.Figure()
 
for tier in ['large', 'medium', 'small']:
    df_t = df_sql[df_sql['tier'] == tier]
    if not df_t.empty:
        fig_sql.add_trace(go.Bar(
            x=df_t['method'],
            y=df_t['rate_raw'],
            name=tier.capitalize()
        ))
 
fig_sql.update_layout(
    barmode='group',
    title='SQL Server Ingestion Performance (Log Scale)',
    xaxis_title='Ingestion Method',
    yaxis_title='Speed (Rows / Second)',
    yaxis_type='log',
    template='plotly_dark',
    xaxis_tickangle=-45
)
fig_sql.show()

Firestore ingestion benchmark

This slice compares the two Firestore write strategies directly so the throughput gap between BulkWriter and manual batch.set() stays visible across all three tiers.

Python example implementing the workflow described above.

fs_methods = [
    'fs_bulkwriter', 'fs_batch'
]
 
df_fs = df_results[df_results['method'].isin(fs_methods)].copy()
df_fs = df_fs.sort_values(['tier_rank', 'rate_raw'], ascending=[True, False])
 
display(df_fs[['method', 'tier', 'rows_fmt', 'elapsed', 'rate']].reset_index(drop=True))
methodtierrows_fmtelapsedrate
0fs_bulkwriterlarge750.0K25m11s496 rows/s
1fs_batchlarge750.0K31m13s400 rows/s
2fs_bulkwritermedium75.0K2m31s497 rows/s
3fs_batchmedium75.0K3m21s372 rows/s
4fs_bulkwritersmall2.5K4.1s604 rows/s
5fs_batchsmall2.5K6.7s372 rows/s

Python example implementing the workflow described above.

fig_fs = go.Figure()
 
for tier in ['large', 'medium', 'small']:
    df_t = df_fs[df_fs['tier'] == tier]
    if not df_t.empty:
        fig_fs.add_trace(go.Bar(
            x=df_t['method'],
            y=df_t['rate_raw'],
            name=tier.capitalize()
        ))
 
fig_fs.update_layout(
    barmode='group',
    title='Firestore Ingestion Performance',
    xaxis_title='Ingestion Method',
    yaxis_title='Speed (Documents / Second)',
    # Linear scale used here as the variance is smaller
    template='plotly_dark',
    xaxis_tickangle=-45
)
fig_fs.show()

Cross-database transfer benchmark

This chart focuses on transfer paths that cross a service boundary, which makes the cost of to_dataframe(), staging, and BulkWriter fan-out easier to compare than in the full benchmark set.

Python example implementing the workflow described above.

transfer_methods = [
    'sql_to_bq_gcs', 'sql_to_bq', 'bq_to_sql', 'sql_to_firestore', 'bq_to_fs'
]
 
df_transfer = df_results[df_results['method'].isin(transfer_methods)].copy()
df_transfer = df_transfer.sort_values(['tier_rank', 'rate_raw'], ascending=[True, False])
 
display(df_transfer[['method', 'tier', 'rows_fmt', 'elapsed', 'rate']].reset_index(drop=True))
methodtierrows_fmtelapsedrate
0sql_to_bq_gcslarge750.0K8m7s1.5K rows/s
1sql_to_bqlarge750.0K8m19s1.5K rows/s
2bq_to_sqllarge750.0K8m34s1.5K rows/s
3bq_to_fslarge750.0K25m52s483 rows/s
4sql_to_firestorelarge750.0K32m51s381 rows/s
5sql_to_bqmedium75.0K51.8s1.4K rows/s
6sql_to_bq_gcsmedium75.0K54.4s1.4K rows/s
7bq_to_sqlmedium75.0K57.4s1.3K rows/s
8bq_to_fsmedium75.0K2m41s466 rows/s
9sql_to_firestoremedium75.0K3m16s383 rows/s
10bq_to_sqlsmall2.5K5.5s451 rows/s
11sql_to_firestoresmall2.5K6.5s387 rows/s
12sql_to_bq_gcssmall2.5K6.5s382 rows/s
13sql_to_bqsmall2.5K7.4s340 rows/s
14bq_to_fssmall2.5K8.9s282 rows/s

Python example implementing the workflow described above.

fig_transfer = go.Figure()
 
for tier in ['large', 'medium', 'small']:
    df_t = df_transfer[df_transfer['tier'] == tier]
    if not df_t.empty:
        fig_transfer.add_trace(go.Bar(
            x=df_t['method'],
            y=df_t['rate_raw'],
            name=tier.capitalize()
        ))
 
fig_transfer.update_layout(
    barmode='group',
    title='Cross-Database Transfer Performance',
    xaxis_title='Transfer Path',
    yaxis_title='Speed (Rows / Second)',
    template='plotly_dark',
    xaxis_tickangle=-45
)

Data exports benchmark

This export view separates extract_table, SQL cursor export, and Firestore document streaming so the final chart reflects outbound throughput rather than ingestion behavior.

Python example implementing the workflow described above.

export_methods = [
    'bq_export_gcs', 'sql_export_csv', 'fs_export_json'
]
 
df_export = df_results[df_results['method'].isin(export_methods)].copy()
df_export = df_export.sort_values(['tier_rank', 'rate_raw'], ascending=[True, False])
 
display(df_export[['method', 'tier', 'rows_fmt', 'elapsed', 'rate']].reset_index(drop=True))
methodtierrows_fmtelapsedrate
0sql_export_csvlarge750.0K6.5s114.7K rows/s
1bq_export_gcslarge750.0K15.8s47.4K rows/s
2fs_export_jsonlarge750.0K2m45s4.5K rows/s
3sql_export_csvmedium75.0K800ms93.7K rows/s
4bq_export_gcsmedium75.0K5.2s14.3K rows/s
5fs_export_jsonmedium75.0K17.0s4.4K rows/s
6sql_export_csvsmall2.5K131ms19.1K rows/s
7fs_export_jsonsmall2.5K707ms3.5K rows/s
8bq_export_gcssmall2.5K4.9s510 rows/s

Python example implementing the workflow described above.

fig_export = go.Figure()
 
for tier in ['large', 'medium', 'small']:
    df_t = df_export[df_export['tier'] == tier]
    if not df_t.empty:
        fig_export.add_trace(go.Bar(
            x=df_t['method'],
            y=df_t['rate_raw'],
            name=tier.capitalize()
        ))
 
fig_export.update_layout(
    barmode='group',
    title='Data Export Performance',
    xaxis_title='Export Method',
    yaxis_title='Speed (Rows / Second)',
    template='plotly_dark',
    xaxis_tickangle=-45
)
fig_export.show()

Cleanup

Removes all staging tables, GCS prefixes, and local export directories created during the benchmark.

pymssql + google-cloud-bigquery — cleanup staging tables

Drops dbo.ohlcv_bench from SQL Server, removes the BigQuery staging and external tables, and purges the exports/ and staging/ GCS prefixes created during the benchmark. Firestore requires no explicit drop — documents are overwritten on the next set() call, so no collection scan or delete is needed.

Python example implementing the workflow described above.

with sql_pymssql() as conn:
    conn.cursor().execute('DROP TABLE IF EXISTS dbo.ohlcv_bench')
    conn.commit()
    print('  SQL Server: ohlcv_bench dropped')
 
bq_client.delete_table(BQ_BENCH_TABLE, not_found_ok=True)
bq_client.delete_table(f'{BQ_BENCH_TABLE}_ext', not_found_ok=True)
print('  BigQuery: ohlcv_bench dropped')
 
print('  Firestore: ohlcv_bench cleared')
 
for blob in gcs_client.list_blobs(BUCKET_NAME, prefix='exports/'):
    blob.delete()
for blob in gcs_client.list_blobs(BUCKET_NAME, prefix='staging/'):
    blob.delete()
print('  GCS: exports/ and staging/ cleaned')
 
if EXPORT_DIR.exists():
    import shutil
    shutil.rmtree(EXPORT_DIR)
    print(f'  Local: exports/ deleted')
print('  Cleanup done')
  SQL Server: ohlcv_bench dropped
  ohlcv_bench dropped
  ohlcv_bench cleared
  exports/ and staging/ cleaned
  Cleanup done

Python Data Ingestion - SQL Server, BigQuery, Firestore Warnings

Never use INSERT in a loop for bulk loads

Row-by-row inserts to SQL Server are 50–200× slower than fast_executemany or BCP. Even for 2.5K rows the difference is measurable; at 750K it is the difference between seconds and minutes.

Correct pattern

Set cursor.fast_executemany = True before executemany(), or use bcp for the fastest local-to-SQL path. For DataFrames, to_sql(..., method='multi', chunksize=5000) is a reasonable middle ground.

Skipping WriteDisposition in BigQuery load jobs causes silent data duplication

The default disposition is WRITE_APPEND. Re-running a notebook cell without truncating first multiplies rows in the target table, corrupting benchmarks.

Correct pattern

Always pass job_config.write_disposition = bigquery.WriteDisposition.WRITE_TRUNCATE in benchmark cells, or wrap loads in an explicit table-delete before each run.

Streaming Firestore documents one by one exhausts your quota and is orders of magnitude slower

Each document.set() call is a separate HTTP request. At 75K documents this is thousands of round-trips and will hit the 1 req/s per-document soft limit.

Correct pattern

Use BulkWriter (preferred) or batch.set() with manual batch.commit() every 500 documents. BulkWriter handles retry and concurrency automatically.

Large GCS-to-SQL transfers that buffer through a DataFrame can exhaust memory

Loading 750K rows from GCS into a pd.DataFrame before writing to SQL Server peaks at 3–4× the raw file size in RAM. On containers with 4 GB limits this causes OOM kills.

Correct pattern

Stream the GCS object as BytesIO, parse in chunks (pd.read_csv(..., chunksize=10000)), and pipe each chunk to fast_executemany without materialising the full DataFrame.

Python Data Ingestion - SQL Server, BigQuery, Firestore Recommendations

  • Always set TQDM_DISABLE=1 at the top of the notebook to suppress pandas_gbq progress bars that pollute benchmark output.
  • Store credentials in environment variables (GOOGLE_APPLICATION_CREDENTIALS, SQL_SERVER_CONN), never hardcoded in notebook cells.
  • Use Parquet as the default interchange format — it is faster to read, smaller on disk, and carries schema metadata that prevents type-coercion errors on load.
  • Run benchmarks in the order small → medium → large (2.5K → 75K → 750K) so early failures do not contaminate timing data for larger tiers.
  • Persist benchmark results to a JSON file after each run so cross-session comparison is possible without re-executing expensive load jobs.
  • For BQ loads from GCS, prefer load_table_from_uri over load_table_from_file for files above 1 GB — GCS-native jobs avoid client-side upload latency.
  • When transferring between services (e.g. BQ → SQL Server), avoid materialising the full result as a DataFrame; stream in pages using the BigQuery Storage Read API or query_job.result().pages.
  • Clean up staging tables, GCS prefixes, and local export directories at the end of each benchmark session to avoid quota and cost surprises.

Python Data Ingestion - SQL Server, BigQuery, Firestore Troubleshooting

Match the failure to the narrowest boundary first: network, schema, write mode, buffer flush, memory, TLS, or region.

pyodbc.OperationalError: Login timeout expired

If pyodbc.OperationalError happens before the first query returns, verify the resolved SQL_SERVER_HOST or GCP_SQL_IP target and confirm that 1433 is reachable from the current runner.

Python example showing the connection target before pyodbc.connect().

import os
 
host = os.environ.get("SQL_SERVER_HOST") or os.environ.get("GCP_SQL_IP", "<missing>")
print(f"sqlserver://{host}:1433")
sqlserver://<missing>:1433

fast_executemany DataError on NULL

If fast_executemany fails on mixed data, normalize None values before executemany() so the driver does not have to infer conflicting SQL types from one column.

Python example normalizing nullable values before bulk insert parameter binding.

values = [1.25, None, 2.00]
normalized = ["" if value is None else f"{value:.2f}" for value in values]
print(normalized)
['1.25', '', '2.00']

notFound: Dataset

A notFound load-job failure usually means the fully qualified project.dataset.table does not exist in the expected project, so re-run the dataset creation step before loading data.

Python example printing the fully qualified BigQuery destination for validation.

project_id = "seclab-dev-ap-26"
dataset_id = "index_data"
table_id = "ohlcv_bench"
print(f"{project_id}.{dataset_id}.{table_id}")
seclab-dev-ap-26.index_data.ohlcv_bench

pandas_gbq appended instead of replacing

If pandas_gbq.to_gbq() leaves too many rows behind, inspect the effective write mode and use if_exists='replace' when the benchmark or ETL run must be idempotent.

Python example making the intended pandas_gbq write mode explicit.

if_exists = "replace"
print(f"pandas_gbq if_exists={if_exists}")
pandas_gbq if_exists=replace

BulkWriter did not flush pending writes

When BulkWriter appears to drop documents, treat bulk_writer.close() as mandatory and put it in finally so buffered writes flush even when earlier work fails.

Python example enforcing a finally flush guard for BulkWriter.

closed = False
try:
    pass
finally:
    closed = True
print(f"bulk_writer_closed={closed}")
bulk_writer_closed=True

GCS download path exhausted memory

If the GCS-to-SQL path runs out of memory, stop materializing the whole file and switch to chunked pd.read_csv(..., chunksize=10000) ingestion so each batch stays bounded.

Python example illustrating chunk-oriented processing instead of one full-frame load.

chunk_sizes = [10000, 10000, 3456]
for index, size in enumerate(chunk_sizes, start=1):
    print(f"chunk {index}: {size} rows")
chunk 1: 10000 rows
chunk 2: 10000 rows
chunk 3: 3456 rows

BCP raised an SSL provider error

When bcp fails during TLS negotiation, test the command line with -C RAW and your field terminator flags first so you can separate certificate-policy issues from CSV parsing issues.

PowerShell example composing the bcp flags that commonly fix TLS negotiation mismatches.

$bcpArgs = @("-C", "RAW", "-t", ",")
$bcpArgs -join " "
-C RAW -t ,

load_table_from_uri stalled on region mismatch

If load_table_from_uri() never finishes, compare the bucket region and dataset region directly because GCS and BigQuery must stay aligned for predictable load-job execution.

Python example checking whether the bucket and dataset regions match.

bucket_region = "europe-west1"
dataset_region = "europe-west1"
print(f"regions_match={bucket_region == dataset_region}")
regions_match=True

Graph View

Backlinks