src/translation/dags/validation_dag.py [58:120]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - create_connection = """data-validation connections add {source_conn_string} && \ data-validation connections add {target_conn_string} && \ data-validation connections list""" # dvt schema validation command string # stores result in .dmt_logs.dmt_dvt_results BQ table (assumes it must be created) schema_validation = "data-validation validate schema --source-conn {source_conn} --target-conn {target_conn} --tables-list {table} --bq-result-handler {project_id}.dmt_logs.dmt_dvt_results -l unique_id={unique_id},table_name={bq_table}" # dvt column validation command string # stores result in .dmt_logs.dmt_dvt_results BQ table (assumes it must be created) column_validation = "data-validation validate column --source-conn {source_conn} --target-conn {target_conn} --tables-list {table} --bq-result-handler {project_id}.dmt_logs.dmt_dvt_results -l unique_id={unique_id},table_name={bq_table}" # dvt row validation command string # stores result in .dmt_logs.dmt_dvt_results BQ table (assumes it must be created) row_validation = "data-validation validate row --source-conn {source_conn} --target-conn {target_conn} --tables-list {table} --bq-result-handler {project_id}.dmt_logs.dmt_dvt_results -l unique_id={unique_id},table_name={bq_table}" copy_sql_files = 'gsutil cp "{source_gcs}/{sql_file}" "source_{sql_file}" & gsutil cp "{target_gcs}/{sql_file}" "target_{sql_file}"' # dvt custom query column type command string # stores result in .dmt_logs.dmt_dvt_results BQ table (assumes it must be created) custom_query_row_validation = 'data-validation validate custom-query row --source-conn {source_conn} --target-conn {target_conn} --source-query-file "source_{sql_file}" --target-query-file "target_{sql_file}" --bq-result-handler {project_id}.dmt_logs.dmt_dvt_results -l unique_id={unique_id},file_name={sql_file}' custom_query_column_validation = 'data-validation validate custom-query column --source-conn {source_conn} --target-conn {target_conn} --source-query-file "source_{sql_file}" --target-query-file "target_{sql_file}" --bq-result-handler {project_id}.dmt_logs.dmt_dvt_results -l unique_id={unique_id},file_name={sql_file}' schema_validation_supported_flags = [ "filter-status", "exclusion-columns", "allow-list", ] column_validation_supported_flags = [ "filter-status", "filters", "count", "sum", "min", "max", "avg", "grouped-columns", "wildcard-include-string-len", "cast-to-bigint", "threshold", ] row_validation_supported_flags = [ "filter-status", "filters", "primary-keys", "hash", "concat", "use-random-row", "random-row-batch-size", "comparison-fields", ] boolean_validation_flags = [ "use-random-row", "wildcard-include-string-len", "cast-to-bigint", ] validation_type_flags_mapping = { "schema": schema_validation_supported_flags, "column": column_validation_supported_flags, "row": row_validation_supported_flags, } - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - src/translation/dvt/main.py [36:96]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - create_connection = """data-validation connections add {source_conn_string} && data-validation connections add {target_conn_string} && data-validation connections list""" # dvt schema validation command string # stores result in .dmt_logs.dmt_dvt_results BQ table (assumes it must be created) schema_validation = "data-validation validate schema --source-conn {source_conn} --target-conn {target_conn} --tables-list {table} --bq-result-handler {project_id}.dmt_logs.dmt_dvt_results -l unique_id={unique_id},table_name={bq_table}" # dvt column validation command string # stores result in .dmt_logs.dmt_dvt_results BQ table (assumes it must be created) column_validation = "data-validation validate column --source-conn {source_conn} --target-conn {target_conn} --tables-list {table} --bq-result-handler {project_id}.dmt_logs.dmt_dvt_results -l unique_id={unique_id},table_name={bq_table}" # dvt row validation command string # stores result in .dmt_logs.dmt_dvt_results BQ table (assumes it must be created) row_validation = "data-validation validate row --source-conn {source_conn} --target-conn {target_conn} --tables-list {table} --bq-result-handler {project_id}.dmt_logs.dmt_dvt_results -l unique_id={unique_id},table_name={bq_table}" copy_sql_files = 'gsutil cp "{source_gcs}/{sql_file}" "source_{sql_file}" & gsutil cp "{target_gcs}/{sql_file}" "target_{sql_file}"' # dvt custom query column type command string # stores result in .dmt_logs.dmt_dvt_results BQ table (assumes it must be created) custom_query_row_validation = 'data-validation validate custom-query row --source-conn {source_conn} --target-conn {target_conn} --source-query-file "source_{sql_file}" --target-query-file "target_{sql_file}" --bq-result-handler {project_id}.dmt_logs.dmt_dvt_results -l unique_id={unique_id},file_name={sql_file}' custom_query_column_validation = 'data-validation validate custom-query column --source-conn {source_conn} --target-conn {target_conn} --source-query-file "source_{sql_file}" --target-query-file "target_{sql_file}" --bq-result-handler {project_id}.dmt_logs.dmt_dvt_results -l unique_id={unique_id},file_name={sql_file}' schema_validation_supported_flags = [ "filter-status", "exclusion-columns", "allow-list", ] column_validation_supported_flags = [ "filter-status", "filters", "count", "sum", "min", "max", "avg", "grouped-columns", "wildcard-include-string-len", "cast-to-bigint", "threshold", ] row_validation_supported_flags = [ "filter-status", "filters", "primary-keys", "hash", "concat", "use-random-row", "random-row-batch-size", "comparison-fields", ] boolean_validation_flags = [ "use-random-row", "wildcard-include-string-len", "cast-to-bigint", ] validation_type_flags_mapping = { "schema": schema_validation_supported_flags, "column": column_validation_supported_flags, "row": row_validation_supported_flags, } - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -