SnowConvert AI: Data validation [Preview]¶

Deployment options¶

The Orchestrator and Workers can be deployed in multiple ways:

• Both on Snowpark Container Services (in the Snowflake account).

• Both in the customer’s environment, including custom hardware, virtual machines, or containers.

• Orchestrator on Snowpark Container Services and Workers in the customer’s environment, or the other way around.

The following requirements apply to the environment where the Orchestrator and Workers run:

• The Orchestrator and Workers are Python packages, so Python 3.11 or higher must be installed.

• Workers typically require an ODBC driver to connect to the source system. For Teradata, the pure-Python teradatasql driver is supported and preferred when installed.

• The Orchestrator must be able to connect to the Snowflake account using a role that has privileges to create the SNOWCONVERT_AI database and create schemas and objects within it.

Schema validation (L1)¶

Schema validation confirms that the structure of each migrated table is preserved in Snowflake. It compares the following attributes between source and target:

• Table name

• Column names

• Ordinal position of each column

• Data types

• Character maximum length for text columns

• Numeric precision and scale for numeric columns

• Row count

Metrics validation (L2)¶

Metrics validation confirms that aggregate statistics of the migrated data match the original source. Specific metrics vary by column data type, but metrics validation typically compares:

• Minimum value

• Maximum value

• Average

• Null count

• Distinct count

• Standard deviation

• Variance

Row validation (L3)¶

Row validation performs row-level or cell-level comparison between source and target. It supports two modes:

• row: compares entire rows identified by the configured index_column_list.

• cell: compares individual cell values and reports the specific columns that differ.

A hybrid mode (two-phase MD5 plus cell drilldown) is also available and is used automatically when the source and target support it, so that only rows flagged by the first phase are re-examined at the cell level. Row validation is disabled by default and is typically applied only to the tables where it is needed because it is the most resource-intensive level.

Installation¶

Install the three Python packages (Python 3.11 or higher) on the machines where the Orchestrator and Workers will run:

pip install snowflake-data-migration-orchestrator
pip install snowflake-data-exchange-agent
pip install snowflake-data-validation

Keep the following in mind:

• The Orchestrator host only needs snowflake-data-migration-orchestrator.

• Workers that will execute Cloud Data Validation tasks need both snowflake-data-exchange-agent and snowflake-data-validation. Workers without snowflake-data-validation can still process Data Migration tasks but skip data_validation tasks.

• On a single host that runs both components, install all three packages in the same Python environment.

Starting the Orchestrator¶

Start the Orchestrator by running:

python -m data_migration_orchestrator start

When invoked without a subcommand, start is assumed for backward compatibility:

python -m data_migration_orchestrator

The start command accepts optional logging flags:

For example, to run with verbose logging to stdout only:

python -m data_migration_orchestrator start \
  --log-destination stdout \
  --log-level DEBUG

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, and data validation metadata is created under SNOWCONVERT_AI.DATA_VALIDATION. To use a different database or schema for that metadata, set the following environment variables before starting the Orchestrator:

• CUSTOM_SNOWFLAKE_DATABASE_FOR_METADATA (default SNOWCONVERT_AI)

• CUSTOM_SNOWFLAKE_SCHEMA_FOR_DATA_MIGRATION_METADATA (default DATA_MIGRATION)

• CUSTOM_SNOWFLAKE_SCHEMA_FOR_DATA_VALIDATION_METADATA (default DATA_VALIDATION)

If you override these values, set the same database and migration 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. Cloud Data Validation Workflows require an active Orchestrator to complete. However, the Orchestrator can be safely stopped at any point and resumed later; ongoing workflows resume at that point.

Starting the Workers¶

The Worker (the data-exchange-agent) provides two subcommands: run (default) and test.

# Start with a configuration file
data-exchange-agent run -c <configuration-file-path>

# Start with the default configuration.toml in the current directory
data-exchange-agent run

# Omitting the subcommand defaults to 'run' (backward compatible)
data-exchange-agent -c <configuration-file-path>

# Custom parallelism and port
data-exchange-agent run --max-parallel-tasks 8 --port 8080

# Task-handling only, without the HTTP server (for multi-worker setups)
data-exchange-agent run --no-server

# Custom base directory for exported files (overrides config)
data-exchange-agent run --local-results-directory /mnt/dea-exports

# Debug mode on a custom port
data-exchange-agent run --debug --port 5001

data-exchange-agent test -c <configuration-file-path>

Creating a Cloud Data Validation Workflow¶

After installing the Orchestrator, create a validation workflow by running the following command:

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

Keep the following in mind:

• The validation configuration specification can be found in Validation workflow configuration reference.

• --source-platform is required. Supported values are sqlserver, redshift, and teradata. The value must match the source_platform field in the JSON configuration file.

• --name is optional. The workflow name must be alphanumeric and cannot start with a digit. Defaults to MY_VALIDATION_WORKFLOW when omitted.

• --connection-name is optional. When omitted, the Orchestrator uses the default Snowflake connection from environment variables. When provided, it should match a named connection in your config.toml or connections.toml file.

For example, to create a named SQL Server validation workflow using a specific Snowflake connection:

python -m data_migration_orchestrator create-data-validation-workflow my-validation.json \
  --source-platform sqlserver \
  --name VALIDATE_SALES_Q1 \
  --connection-name my-snowflake

New workflow rows are inserted into the WORKFLOW table in the data migration metadata schema (default SNOWCONVERT_AI.DATA_MIGRATION) with WORKFLOW_TYPE set to data-validation. Validation results and related objects are stored under the data validation metadata schema (default SNOWCONVERT_AI.DATA_VALIDATION).

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

Monitoring a Cloud Data Validation 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 haven’t reached a terminal state (COMPLETED or FAILED).

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

In the data validation metadata schema (default SNOWCONVERT_AI.DATA_VALIDATION), the following views can be queried to understand the status of validation workflows:

In the same schema, the DATA_VALIDATION_DASHBOARD Streamlit dashboard provides a visual overview of validation progress and results, including a Table Progress tab that aggregates the views above.

You can also inspect validation queries in QUERY_HISTORY using the QUERY_TAG values set by the Orchestrator and Workers. See Query tagging.

Validation outcomes are classified into three categories:

Connecting to Snowflake with a PAT¶

Use Programmatic Access Tokens (PATs) for connections used by the Orchestrator and Workers. This avoids the need to constantly authenticate through the browser or with an authenticator app. You need to establish a Network Policy or temporarily bypass the requirement for a Network Policy (which you can do from Snowsight).

Running Orchestrator and Workers on SPCS¶

To leverage Snowflake compute for these tasks:

1. Prepare Docker images that use the Python modules with the appropriate configuration.

2. Push those Docker images to an Image Repository in Snowflake.

3. Execute the Orchestrator and/or Worker images using Snowpark Container Services.

Keep the following in mind:

• Run them as Services, not Jobs.

• You can run only one component (Orchestrator or Workers) in SPCS and the other on a different platform.

• Monitor the SPCS service and suspend it when it isn’t being used.

• Depending on the network configuration of the source system, you might need to configure an External Access Integration so that these services can connect to your source system.

Initial testing¶

For an early test run, use a separate validation configuration whose tables array lists only the table or small set of tables you want to validate. On each of those entries:

• Set where_clause (and optionally target_where_clause) to an SQL-like predicate so only a subset of rows is validated, for example, a bounded primary-key range or a narrow date range in the source dialect.

• Keep validation_configuration.row_validation disabled for the first run, and enable it later on a smaller subset of tables.

• Use a small target_partition_size_mb or target_partition_size_rows to keep partitions tiny during the test.

After you confirm connectivity and results, create your full workflow: remove or relax where_clause, adjust partition sizes, and enable row validation only on the tables where it is needed.

Managing Workers¶

The time it takes to complete a validation workflow depends on many variables, but the number of Workers (and threads per Worker) has the greatest impact, as it determines how many validation tasks can be executed in parallel. Consider the following:

• You don’t need to run two Workers on the same machine. If you want more parallelism on a single machine, increase the thread count instead (--max-parallel-tasks on data-exchange-agent run).

• Network bandwidth greatly affects Worker speed and is shared between threads of a Worker.

• Even with many Workers and threads running in parallel, the source system might not have enough resources to handle the load.

• Keep a low Worker count to avoid overloading your source system.

• Consider stopping some or all Workers when the source system is already under heavy load from other operations.

Query tagging¶

Both the Orchestrator and the Worker automatically set Snowflake’s QUERY_TAG session parameter on every query they submit. Tags are compact JSON strings containing identifiers such as the workflow ID, task ID, and component version. You can use these tags to filter and attribute validation queries in QUERY_HISTORY:

SELECT query_text, query_tag, start_time
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE TRY_PARSE_JSON(query_tag):DMVF_WORKFLOW_ID IS NOT NULL
ORDER BY start_time DESC;

Validation workflow configuration reference¶

The validation workflow configuration file is a JSON object. The following sections describe its structure and properties.

{
  "source_platform": "sqlserver",
  "database_mappings": {
    "SampleStoreDB": "samplestoredb"
  },
  "schema_mappings": {
    "data_migration_cloud_test": "data_migration_cloud_test"
  },
  "validation_configuration": {
    "schema_validation": true,
    "metrics_validation": true,
    "row_validation": false
  },
  "comparison_configuration": {
    "tolerance": 0.001
  },
  "tables": [
    {
      "fully_qualified_name": "SampleStoreDB.data_migration_cloud_test.store_employee",
      "target_name": "target_employee",
      "column_names_to_partition_by": ["ID"]
    },
    {
      "fully_qualified_name": "SampleStoreDB.data_migration_cloud_test.Sales_Simple",
      "column_names_to_partition_by": ["ID"],
      "index_column_list": ["ID"],
      "target_partition_size_mb": 256,
      "validation_configuration": {
        "row_validation": true,
        "row_validation_mode": "cell",
        "max_failed_rows_number": 500
      },
      "where_clause": "is_deleted = 0"
    }
  ]
}

{
  "source_platform": "redshift",
  "target_database": "TARGET_DB",
  "schema_mappings": {
    "ecommerce_raw": "ecommerce_raw"
  },
  "validation_configuration": {
    "schema_validation": true,
    "metrics_validation": true,
    "row_validation": false
  },
  "tables": [
    {
      "fully_qualified_name": "snowconvert_demo.ecommerce_raw.customers",
      "column_names_to_partition_by": ["customer_id"]
    },
    {
      "fully_qualified_name": "snowconvert_demo.ecommerce_raw.orders",
      "column_names_to_partition_by": ["order_id"],
      "index_column_list": ["order_id"],
      "validation_configuration": {
        "row_validation": true,
        "row_validation_mode": "row"
      }
    }
  ]
}

{
  "source_platform": "teradata",
  "target_database": "TARGET_DB",
  "validation_configuration": {
    "schema_validation": true,
    "metrics_validation": true,
    "row_validation": false
  },
  "comparison_configuration": {
    "tolerance": 0.0001
  },
  "tables": [
    {
      "fully_qualified_name": "tpcds.store_sales",
      "column_names_to_partition_by": ["ss_ticket_number"],
      "target_partition_size_mb": 512
    },
    {
      "fully_qualified_name": "tpcds.customer",
      "target_schema": "PUBLIC",
      "column_names_to_partition_by": ["c_customer_sk"],
      "validation_configuration": {
        "row_validation": true,
        "row_validation_mode": "cell"
      }
    }
  ]
}

Worker configuration¶

The Worker configuration file uses TOML format and is shared between Data Migration and Cloud Data Validation. Workers that execute validation tasks need the snowflake-data-validation package installed in the same Python environment.

When selected_task_source is "snowflake_stored_procedure", the Worker calls task-queue stored procedures using application.snowflake_database_for_metadata and application.snowflake_schema_for_data_migration_metadata. These values are independent of the Snowflake session defaults (SNOWFLAKE_DATABASE, SNOWFLAKE_SCHEMA) in the connection profile.

An example configuration file looks like this:

selected_task_source = "snowflake_stored_procedure"

[application]
max_parallel_tasks = 4
task_fetch_interval = 30
# Optional: only if the Orchestrator uses CUSTOM_SNOWFLAKE_* overrides for metadata location
# snowflake_database_for_metadata = "SNOWCONVERT_AI"
# snowflake_schema_for_data_migration_metadata = "DATA_MIGRATION"

# SQL Server connection (standard authentication)
[connections.source.sqlserver]
username = "username"
password = "password"
database = "database_name"
host = "127.0.0.1"
port = 1433

# Snowflake target connection
[connections.target.snowflake_connection_name]
connection_name = "connection_name"

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

[connections.source.sqlserver]
odbc_driver = "ODBC Driver 17 for SQL Server"

[connections.source.sqlserver]
username = "sa"
password = "mypassword"
database = "mydb"
host = "my-server.example.com"
port = 1433
encrypt = true
trust_server_certificate = false

[connections.source.redshift]
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"

[connections.source.redshift]
username = "myuser"
password = "mypassword"
database = "mydatabase"
host = "my-cluster.abcdef123456.us-west-2.redshift.amazonaws.com"
port = 5439
auth_method = "standard"

[connections.source.teradata]
host = "your-teradata-host.example.com"
port = 1025
database = "tpcds"
username = "your_username"
password = "your_password"
# driver_name = "Teradata Database ODBC Driver 17.20"  # only needed for ODBC fallback
# dbc_name = "TDPID_ALIAS"  # optional; defaults to host