Great question — you can stream from SQL Server to Snowflake with very low memory and zero local disk by combining SQLAlchemy’s server-side cursors with chunked writes to Snowflake. Below are two proven patterns in Prefect 3.x using your existing

SqlAlchemyConnector

block. Assumptions: - You’re on Prefect 3.x - You already have a SqlAlchemyConnector block for SQL Server - You can create/use a SnowflakeConnector block (shown below if you don’t have one yet) Option A (recommended first): Stream SQL Server -> chunked DataFrames -> write_pandas to Snowflake - Very fast (Arrow-optimized) - Memory bounded by chunk size (O(batch_size)) - No local files

from prefect import flow, task
from prefect_sqlalchemy import SqlAlchemyConnector
from prefect_snowflake.database import SnowflakeConnector
from sqlalchemy import text
from snowflake.connector.pandas_tools import write_pandas
import pandas as pd


def stream_sqlserver_as_dfs(sql_block_name: str, query: str, batch_size: int = 50_000):
    """
    Stream rows out of SQL Server using server-side cursors and yield pandas DataFrames.
    Memory stays bounded by batch_size; no local files used.
    """
    sql = SqlAlchemyConnector.load(sql_block_name)
    with sql.get_connection(begin=False) as conn:
        result = conn.execution_options(stream_results=True).execute(text(query))
        cols = result.keys()
        batch = []
        for row in result:
            batch.append(tuple(row))
            if len(batch) >= batch_size:
                yield pd.DataFrame(batch, columns=cols)
                batch = []
        if batch:
            yield pd.DataFrame(batch, columns=cols)


@task(retries=3, retry_delay_seconds=30)
def append_df_to_snowflake(sf_block_name: str, df: pd.DataFrame, table: str) -> int:
    """
    Chunked, Arrow-accelerated write to Snowflake.
    """
    sf = SnowflakeConnector.load(sf_block_name)
    with sf.get_connection() as conn:
        success, nchunks, nrows, _ = write_pandas(
            conn,
            df,
            table_name=table,
            database=sf.database,
            schema=sf.schema_,
        )
    if not success:
        raise RuntimeError("write_pandas reported failure")
    return nrows


@flow(name="sqlserver-to-snowflake-streaming")
def migrate_sqlserver_to_snowflake(
    sql_block_name: str = "sqlserver-block",
    sf_block_name: str = "snowflake-connector",
    source_query: str = "SELECT * FROM dbo.large_table ORDER BY id",
    target_table: str = "STAGING.LARGE_TABLE",
    batch_size: int = 50_000,
):
    total = 0
    for df in stream_sqlserver_as_dfs(sql_block_name, source_query, batch_size):
        total += append_df_to_snowflake(sf_block_name, df, target_table)
    return {"rows_loaded": total}

Tuning tips for Option A - Start with

batch_size

25k–100k depending on row width and available RAM (e.g., 50k is a solid default). - Use

ORDER BY

on a stable key to make resuming idempotent (log last max ID). - Size your Snowflake warehouse appropriately for bulk loads. Option B (largest volumes, zero local disk): Stream -> write CSV chunks to cloud storage -> Snowflake COPY INTO - Ideal when moving 100s of GB+ or when you want resilient, restartable bulk loads - Uses virtually no local disk; memory bounded by chunk size - Requires an external stage (e.g., S3/Azure/GCS) or internal stage with files uploaded via cloud SDK Example with S3 external stage (no local files): ``` import io, csv, gzip, boto3 from prefect import flow, task from prefect_sqlalchemy import SqlAlchemyConnector from prefect_snowflake.database import SnowflakeConnector, snowflake_query from sqlalchemy import text def stream_to_s3_chunks(sql_block_name, query, bucket, prefix, batch_size=100_000, gzip_compress=True): sql = SqlAlchemyConnector.load(sql_block_name) s3 = boto3.client("s3") with sql.get_connection(begin=False) as conn: result = conn.execution_options(stream_results=True).execute(text(query))

cols = result.keys() batch, part = [], 0 for row in result: batch.append(tuple(row)) if len(batch) >= batch_size: buf = io.BytesIO() if gzip_compress: gz = gzip.GzipFile(fileobj=buf, mode="wb") writer = csv.writer(io.TextIOWrapper(gz, encoding="utf-8", newline="")) else: writer = csv.writer(io.TextIOWrapper(buf, encoding="utf-8", newline="")) writer.writerow(cols) writer.writerows(batch) if gzip_compress: writer = None; gz.close() buf.seek(0) key = f"{prefix}/part-{part:06d}.csv" + (".gz" if gzip_compress else "") s3.upload_fileobj(buf, bucket, key) part += 1 batch = [] if batch: buf = io.BytesIO() if gzip_compress: gz = gzip.GzipFile(fileobj=buf, mode="wb") writer = csv.writer(io.TextIOWrapper(gz, encoding="utf-8", newline="")) else: writer = csv.writer(io.TextIOWrapper(buf, encoding="utf-8", newline="")) writer.writerow(cols) writer.writerows(batch) if gzip_compress: writer = None; gz.close() buf.seek(0) key = f"{prefix}/part-{part:06d}.csv" + (".gz" if gzip_compress else "") s3.upload_fileobj(buf, bucket, key) @task def snowflake_copy_from_stage(sf_block_name: str, table: str, stage: str, prefix: str, gz: bool = True): sf = SnowflakeConnector.load(sf_block_name) compression = "GZIP" if gz else "NONE" copy_sql = f""" COPY INTO {table} FROM @{stage}/{prefix}/ FILE_FORMAT=(TYPE=CSV SKIP_HEADER=1 FIELD_OPTIONALLY_ENCLOSED_BY='\"' COMPRESSION={compression}) PATTERN='.*\\.csv(\\.gz)?' """ return snowflake_query(copy_sql, snowflake_connector=sf) @flow(name="sqlserver-to-snowflake-via-stage") def migrate_via_external_stage( sql_block_name: str = "sqlserver-block", sf_block_name: str = "snowflake-connector", source_query: str = "SELECT * FROM dbo.large_table ORDER BY id", target_table: str = "STAGING.LARGE_TABLE", s3_bucket: str = "my-etl-bucket", s3_prefix: str = "loads/my_table/run_2025_01_01", stage_name: str = "MY_S3_STAGE", # pre-created Snowflake stage pointing at the bucket batch_size: int = 100_000, gzip_compress: bool = True, ): stream_to_s3_chunks(sql_block_name, source_query, s3_bucket, s3_prefix, batch_size, gzip_compress) return snowflake_copy_from_stage(sf_block_name, target_table, stage_name, s3_prefix, gzip_compress)

Why these work well
- SQL Server reading is truly streamed via `execution_options(stream_results=True)`, so you never materialize the full result in memory.
- Option A uses Snowflake’s Arrow-based `write_pandas` path for fast, chunked inserts with no local files.
- Option B pushes chunks straight to object storage and uses Snowflake’s highly optimized `COPY INTO`, which is resilient and restartable.

Optional: creating the SnowflakeConnector block (if you don’t have one yet)

from prefect_snowflake.credentials import SnowflakeCredentials from prefect_snowflake.database import SnowflakeConnector creds = SnowflakeCredentials( account="abc12345.us-east-1", user="etl_user", password="***", role="TRANSFORMER", ) creds.save("sf-creds", overwrite=True) sf = SnowflakeConnector( credentials=creds, database="ANALYTICS", schema_="STAGING", warehouse="COMPUTE_WH", ) sf.save("snowflake-connector", overwrite=True) ``` Quick tuning checklist - Use

ORDER BY

and log the last processed key to resume safely. - Start with 25k–100k row batches; reduce for very wide rows. - Size your Snowflake warehouse up during the load; scale down after. - Add retries to the write tasks; Snowflake COPY is already resilient.

- If you need strict zero local disk, prefer Option B. Questions to tailor this for you - Rough size of the dataset (rows/GB)? - Do you have a Snowflake stage available (S3/Azure/GCS)? If so, which cloud? - Do you need any transforms during the move, or is it straight copy? - Any constraints on runtime or memory we should target? If you share those, I can tighten batch sizing and the exact pattern.