SnowConvert AI: Data migration [Preview]¶

Starting the Orchestrator¶

After installing the Orchestrator, start it by running the following command:

python -m data_migration_orchestrator

Before running, make sure the SNOWFLAKE_CONNECTION_NAME environment variable is set to a value that matches one of the connection names in your Snowflake config.toml or connections.toml. That is the name of the connection used to connect to the target Snowflake account.

By default, workflow and task metadata objects are created under SNOWCONVERT_AI.DATA_MIGRATION. To use a different database or schema for that metadata, set the environment variables CUSTOM_SNOWFLAKE_DATABASE_FOR_METADATA (default SNOWCONVERT_AI) and CUSTOM_SNOWFLAKE_SCHEMA_FOR_DATA_MIGRATION_METADATA (default DATA_MIGRATION) before starting the Orchestrator. If you override these values, set the same database and schema in each Worker’s configuration using snowflake_database_for_metadata and snowflake_schema_for_data_migration_metadata under [application] (see Worker configuration).

The Orchestrator runs until you stop it. Data Migration Workflows require an active Orchestrator to complete. However, the Orchestrator can be safely stopped at any point and resumed later; ongoing Data Migration Workflows are resumed at that point.

Starting the Workers¶

After installing the Worker, start it by running the following command:

python -m data_exchange_agent -c <configuration-file-path>

The path to the configuration file can be omitted. In that case, the Worker looks for a file called configuration.toml in your current directory. For the Worker configuration specification, see Worker configuration.

Workers run until you stop them. Data Migration Workflows require at least one active Worker to complete. However, Workers can be safely stopped at any point and resumed later; ongoing Data Migration Workflows are resumed at that point.

Creating a Data Migration Workflow¶

After installing the Orchestrator, create workflows by running the following command:

python -m data_migration_orchestrator create-data-migration-workflow <workflow-config-file-path> \
  --name <workflow-name> \
  --connection-name <connection-name> \
  --source-platform <source-platform>

Keep the following in mind:

• The Workflow Configuration specification can be found in Workflow configuration reference.

• The workflow name must be composed of alphanumeric characters and cannot start with a digit.

• You must specify the name of the Snowflake connection you want to use, as it appears in your config.toml or connections.toml file.

• Supported source platforms are sqlserver and redshift.

Monitoring a Data Migration Workflow¶

Each workflow goes through different stages throughout its lifecycle:

• Pending: No tasks have been created for this workflow yet.

• Executing: Tasks have been created for this workflow and there are still tasks that have not reached a terminal state (COMPLETED or FAILED).

• Completed: All tasks have reached a terminal state (COMPLETED or FAILED).

In the SNOWCONVERT_AI.DATA_MIGRATION schema, the following tables and views can be queried to understand the status of one or more workflows:

• WORKFLOW: Contains one row per workflow, including start/end time, status, and configuration.

• TABLE_PROGRESS_WITH_EXAMPLE_ERROR: Contains one row per table being migrated as part of a workflow. Includes information about how many partitions are in each stage (extraction, loading, completed, or failed), as well as related errors. Can be filtered by WORKFLOW_ID.

• DATA_MIGRATION_ERROR: For each partition of a table being migrated, contains the first known error that affected the migration of that partition. Can be filtered by WORKFLOW_ID.

In the same schema, the DATA_MIGRATION_DASHBOARD Streamlit dashboard can be used to monitor workflows. This dashboard presents data from those tables and views.

Top-level object¶

When defaultTableConfiguration is present, each object in tables is merged with those defaults: shared fields apply to every table unless the same field is set again on a specific table entry, in which case the table-level value wins.

TableConfiguration model¶

Defines the settings for migrating a single table.

SourceTargetIdentifier model¶

A nested object used within TableConfiguration to specify a database object. For source, use only the properties in the following table. For target, you can also set the optional properties in Additional target properties.

Additional target properties¶

The following optional fields apply only to the target object (not to source).

Iceberg configuration (target.icebergConfig)¶

Used when target.tableType is "iceberg". Account setup (external volumes, catalog integrations, stages, and privileges) follows Snowflake’s Iceberg documentation; see Apache Iceberg™ tables, Create an Iceberg table, and Configure an external volume.

Partition size (partitionSize)¶

Controls how large each partition should be during extraction. You can use a string or an object.

When you use the object form, specify only one of targetSizeMb or maxRowsPerPartition.

auto (default):

"partitionSize": "auto"

Target size in MB:

"partitionSize": { "targetSizeMb": 2048 }

Maximum rows per partition:

"partitionSize": { "maxRowsPerPartition": 500000 }

ColumnTypeMapping model¶

A nested object used within TableConfiguration to specify type mappings for a column.

ColumnNameMapping model¶

A nested object used within TableConfiguration to specify column name mappings.

ExtractionStrategy model¶

Configures the method for data extraction.

Extraction: regular (Default)

"extraction": {
  "strategy": "regular"
}

Extraction: unload (Redshift only)

"extraction": {
  "strategy": "unload",
  "externalStage": "MY_DB.MY_SCHEMA.S3_EXTERNAL_STAGE"
}

SynchronizationStrategy model¶

Configures the approach for incremental data syncing on subsequent runs.

Strategy: none (Default)

Performs a full extraction of all partition data on every run. No synchronization metadata is stored.

"synchronization": {
  "strategy": "none"
}

Use when data is small, changes are unpredictable, or guaranteed consistency is needed.

Strategy: checksum

Computes a hash of all column values for each partition on the source. Only changed partitions are cleared and re-extracted in the target.

"synchronization": {
  "strategy": "checksum"
}

Use when you need to detect any change in a partition but lack a reliable monotonic column (for example, dimension tables). Note that this requires a checksum computation on the source for every partition on every run.

Strategy: watermark

Tracks a monotonic column (timestamp, ID, or version) to sync only rows where the watermark value is greater than the maximum observed in the previous sync.

"synchronization": {
  "strategy": "watermark",
  "watermarkColumn": "UPDATED_AT"
}

Use when your table has a reliable monotonic column that increases on insert/update (for example, fact tables or event logs).

Example workflow: Redshift UNLOAD with Iceberg targets¶

The following workflow excerpt combines Redshift UNLOAD with Iceberg table targets, including Snowflake-managed Iceberg defaults and a per-table external catalog override:

{
  "defaultTableConfiguration": {
    "source": {
      "schemaName": "public",
      "databaseName": "analytics_db"
    },
    "target": {
      "schemaName": "public",
      "databaseName": "TARGET_DB",
      "tableType": "iceberg",
      "icebergConfig": {
        "catalog": "SNOWFLAKE",
        "externalVolume": "my_iceberg_ext_vol",
        "baseLocationPrefix": "migrations/redshift",
        "sourceDataStage": "@TARGET_DB.PUBLIC.ICEBERG_SOURCE_STAGE"
      }
    },
    "extraction": {
      "strategy": "unload",
      "externalStage": "TARGET_DB.PUBLIC.S3_EXTERNAL_STAGE"
    },
    "partitionSize": "auto"
  },
  "tables": [
    {
      "source": { "tableName": "customers" },
      "target": { "tableName": "customers" },
      "columnNamesToPartitionBy": ["customer_id"]
    },
    {
      "source": { "tableName": "events" },
      "target": {
        "tableName": "events",
        "tableType": "iceberg",
        "icebergConfig": {
          "catalog": "my_glue_catalog_integration",
          "externalVolume": "my_iceberg_ext_vol",
          "catalogTableName": "glue_db.events"
        }
      },
      "columnNamesToPartitionBy": ["event_id"]
    }
  ]
}

Affinity¶

By specifying an affinity for a Workflow, you are indicating that you want specific workers to help with the execution of that Workflow. This can be particularly useful in cases in which you want to have some workers extract from one source and other workers extract from a different source. The rules for matching workers with tasks are:

• A task without affinity will be picked up by any worker, independently of the worker’s affinity.

• A worker without affinity will pick up any task, independently of the task’s affinity.

• A task with a given affinity will not be picked up by a worker with different affinity.

Affinity only needs to be a String; its format is defined by the user. For example, all of these are valid: sql-server, DEV_SERVER, my_custom_server, ::blue::.

Source connection configuration examples¶

The following examples show the three main source connection configurations:

1. SQL Server (Standard Authentication)

[connections.source.sqlserver]
username = "username"
password = "password"
database = "database_name"
host = "127.0.0.1"
port = 1433

2. Amazon Redshift (IAM Authentication)

[connections.source.redshift-iam]
username = "demo-user"
database = "demo_db"
auth_method = "iam-provisioned-cluster"
cluster_id = "my-aws-cluster"
region = "us-west-2"
access_key_id = "your-access-key-id"
secret_access_key = "your-secret-access-key"
# Optional fields for UNLOAD strategy
# unload_s3_bucket = "my-migrations-bucket"
# unload_iam_role_arn = "arn:aws:iam::123456789012:role/MyRole"

3. Amazon Redshift (Standard Authentication)

[connections.source.redshift-standard]
username = "myuser"
password = "mypassword"
database = "mydatabase"
host = "my-cluster.abcdef123456.us-west-2.redshift.amazonaws.com"
port = 5439
auth_method = "standard"
# Optional fields for UNLOAD strategy
# unload_s3_bucket = "my-migrations-bucket"
# unload_iam_role_arn = "arn:aws:iam::123456789012:role/MyRole"

Create a stage integration to S3¶

If you don’t have an existing stage configured, you need to create a Snowflake external stage that integrates with your S3 bucket. You can create the stage using the following SQL command in Snowflake:

CREATE OR REPLACE STAGE <stage_name>
  URL = 's3://<your_bucket_name>/<path>/'
  STORAGE_INTEGRATION = <your_storage_integration>
  FILE_FORMAT = (TYPE = 'PARQUET');

Alternatively, if you’re using AWS credentials directly:

CREATE OR REPLACE STAGE <stage_name>
  URL = 's3://<your_bucket_name>/<path>/'
  CREDENTIALS = (AWS_KEY_ID = '<your_aws_key_id>' AWS_SECRET_KEY = '<your_aws_secret_key>')
  FILE_FORMAT = (TYPE = 'PARQUET');

Replace the placeholders:

• <stage_name>: Your desired stage name (for example, my_redshift_stage)

• <your_bucket_name>: Your S3 bucket name

• <path>: Optional path within the bucket

• <your_storage_integration>: Your Snowflake storage integration name (recommended method)

• <your_aws_key_id> and <your_aws_secret_key>: Your AWS IAM user credentials (if not using storage integration)

Verify stage integration¶

After setting up your stage, verify that the integration is working correctly before proceeding with data migration. You can verify the stage integration by running the following command in Snowflake:

LIST @<stage_name>;

This command should execute successfully without errors. If the stage is newly created and empty, it may return no results, which is expected.

To perform a more thorough verification, you can test the stage by uploading a test file:

PUT file:///<local_test_file_path> @<stage_name>;
LIST @<stage_name>;

If the commands execute successfully and you can see the uploaded file, your stage integration is configured correctly.

DESCRIBE STAGE <stage_name>;