mkpipe-extractor-snowflake

Snowflake extractor plugin for MkPipe. Reads Snowflake tables using the native Snowflake Spark connector (spark-snowflake), which transfers data via internal staging (S3/Azure/GCS) — significantly faster than JDBC for large datasets.

Documentation

For more detailed documentation, please visit the GitHub repository.

License

This project is licensed under the Apache 2.0 License - see the LICENSE file for details.

Connection Configuration

connections:
  snowflake_source:
    variant: snowflake
    host: myaccount.snowflakecomputing.com
    port: 443
    database: MY_DATABASE
    schema: MY_SCHEMA
    user: myuser
    password: mypassword
    warehouse: MY_WAREHOUSE

With RSA key pair authentication:

connections:
  snowflake_source:
    variant: snowflake
    host: myaccount.snowflakecomputing.com
    port: 443
    database: MY_DATABASE
    schema: MY_SCHEMA
    user: myuser
    warehouse: MY_WAREHOUSE
    private_key_file: /path/to/rsa_key.p8
    private_key_file_pwd: mypassphrase   # omit if key is unencrypted

Table Configuration

pipelines:
  - name: snowflake_to_pg
    source: snowflake_source
    destination: pg_target
    tables:
      - name: MY_SCHEMA.EVENTS
        target_name: stg_events
        replication_method: full
        fetchsize: 100000

Incremental Replication

      - name: MY_SCHEMA.EVENTS
        target_name: stg_events
        replication_method: incremental
        iterate_column: UPDATED_AT
        iterate_column_type: datetime
        partitions_column: ID
        partitions_count: 8
        fetchsize: 50000

Custom SQL

      - name: MY_SCHEMA.EVENTS
        target_name: stg_events
        replication_method: full
        custom_query: "SELECT ID, USER_ID, EVENT_TYPE, CREATED_AT FROM MY_SCHEMA.EVENTS WHERE {query_filter}"

Use {query_filter} as a placeholder — it is replaced with the incremental WHERE clause on incremental runs, or WHERE 1=1 on full runs.

Read Parallelism

Snowflake extractor uses the native Snowflake Spark connector with Spark's partition support. For large tables, set partitions_column and partitions_count to read in parallel:

      - name: MY_SCHEMA.EVENTS
        target_name: stg_events
        replication_method: incremental
        iterate_column: UPDATED_AT
        iterate_column_type: datetime
        partitions_column: ID       # numeric column to split on
        partitions_count: 8         # number of parallel Spark partitions
        fetchsize: 50000

How it works

• Spark reads the min/max of partitions_column and divides the range into partitions_count equal slices
• Each slice is fetched by a separate Spark task via the Snowflake connector
• fetchsize controls how many rows are fetched per round-trip

Performance Notes

• Full replication: partitioning is not applied (only works with incremental).
• partitions_column should be a numeric column with good distribution (e.g. surrogate key).
• Snowflake Warehouse size is the primary performance lever — a larger warehouse processes queries faster regardless of JDBC partitioning.
• Keep fetchsize high (50,000–200,000) to minimize round-trips; Snowflake handles large result sets efficiently.

All Table Parameters