Big part of Data Warehouse development has always been tied to structuring data acquisition pipelines before more rigid modelling takes place and data is wrangled (most likely) using the most the venerable methodologies today: Data Vault, Kimbal or Inmon. I have covered data acquisition topic in a number of previous posts (mainly HERE<\/a> and HERE<\/a>), but would like to further expand on this subject with a short post on how to enhance this process (using a simplified example) and enable hash-partitioning on individual tables. Having concurrent executions i.e. a number of loads running in parallel, one for every source table, is fairly straightforward to achieve using SQL Server platform and a little bit of T-SQL. There are certainly more elegant approaches which enable turning sequential loads into a parallel ones, but with a little bit of elbow grease one can script out a batch of good, old-fashioned SQL Server Agent jobs with ease to be spun up and run simultaneously (example HERE<\/a>). However, in some cases, source tables can be quite large and having a single transaction responsible for the insertion of the whole table\u2019s content into a target object can be quite time-consuming. The situation can be further exacerbated if the primary key on the source table is a composite one or not of a numeric type.<\/p>\n One of the clients I worked with a few years ago had this problem. They needed to reload the entire content of a few source tables on a nightly schedule and the framework they used was up to the task until data stored in those objects grew by a large margin. Those tables contained a lot of text data i.e. email, chat and telephone transcripts so many of the fields were quite wide. As the acquisition framework they used dynamically generated BIML scrips and SSIS packages, there was very little one could do to tune or modify the logic due to the proprietary nature of the framework\u2019s design. Data volumes increased exponentially and over time packages responsible for objects with very wide attributes started timing out, requiring expensive, bespoke development. Another, much more recent case, involved my current employer\u2019s choice of survey software. If you\u2019ve ever used LimeSurvey<\/a>, you will know that each survey\u2019s data is stored in a single, very wide table (think hundreds of attributes), with a dedicated column assigned to each question. That\u2019s not a big issue, especially that MySQL (database used in LimeSurvey) has a row limit of 65,535 bytes, however, this schema architecture enabled objects with very wide rows and shallow row count to be proliferated across the database. Extracting this data, particularly when lots of textual values are involved, can be quite slow when thousands of tables are involved.<\/p>\n To demonstrate how to mitigate this type of problem, I will create a simple scenario where a source system table acquisition job can be run across multiple, hash-partitioned ‘chunks’ concurrently. Based on a distinct primary (single column or composite) key, we will be able to distribute the table content across multiple, consistent and non-blocking streams of data, thus providing a significant acceleration of data movement and cutting the time dedicated to ‘E’ (out of ELT\/ETL) by a large margin. I will also compare the time it takes to load this data using variable number of streams i.e. between 2 and 8, to demonstrate how this framework can be tuned to enable accelerated performance.<\/p>\n The framework\u2019s simplified architecture view can be depicted as per the diagram below.<\/p>\n On the left we have our source system (augmented by the system metadata stored in a separate database) and a ‘parent’ process used for jobs coordination and dispatching. In the middle we have individual ‘workers’ operating on a hash-defined partition of source data. Finally, on the right we have our target system where it all gets assembled into a mirror copy of the source table. I will go over how all this ‘hangs together’ in more details in the section below.<\/p>\n First, let\u2019s create all required resources on Azure. The following Python script is used to generate Azure resource group, Azure SQL server and an empty database. You can easily replace it with a different self-hosted SQL Server instance (it doesn’t need to live in Azure).<\/p>\n Providing all necessary Azure SDK for Python modules were installed and referenced correctly and we’ve signed in using Azure CLI, we should see the following status output at the end of script execution.<\/p>\n Next, let\u2019s generate some mock data. This is required to simulate the scenario I was referring to before i.e. a table with a large number of columns and variable-length text data which we will try to acquire as efficiently and quickly as possible. The following script creates a small stored procedure used to generate dummy data, Dummy_Table object and finally, it assigns ID column as a primary key on the newly populated table.<\/p>\n The stored procedure (code below) is configured to create 300 columns across 100000 rows of dense, text data. When finished (executed for approx. 20min with the default configuration of data volume and Azure resources specified in the script above), the very wide table schema as well as synthetically created data will look similar to the one in the image below (click on image to enlarge).<\/p>\n Next, I will create a linked server connection from my on-premise SQL Server instance to the one I\u2019ve just provisioned in Azure and recreate the source Dummy_Table schema in the target database.<\/p>\n Finally, in this example I will also be useing two small databases (EDW_Control and AdminDBA) which contain metadata information on the object(s) I will be processing and error logs for loads which fail to run correctly. The structure of EDW_Control database is pretty straightforward – it uses a collection of tables to track things like source and target schema names, record counts, table and indexes size etc. The important part is the fields called \u2018Is_Big\u2019 and \u2018ETL_Batch_No\u2019 which dictate whether the nominated object should be hash-partitioned and if so, into how many batches\/streams. I wrote more about how these tables are structured in one of my previous blog post HERE<\/a>. Details on how to build and deploy AdminDBA database can be found in my previous posts HERE<\/a> and HERE<\/a>. For reference, the EDW_Control database schema and a snapshot of metadata I will be processing looks as per below.<\/p>\n![\"\"](\"http:\/\/bicortex.com\/bicortex\/wp-content\/post_content\/\/2021\/06\/Hash_Partitioned_Data_Sync_Architecture.png\")
<\/a><\/p>\nAzure Source Database and Dummy Data Setup<\/h3>\n
\r\nfrom azure.common.client_factory import get_client_from_cli_profile\r\nfrom azure.mgmt.resource import ResourceManagementClient\r\nfrom azure.mgmt.sql import SqlManagementClient\r\nfrom msrestazure.azure_exceptions import CloudError\r\nfrom os import popen\r\nimport pyodbc\r\n\r\nRESOURCE_GROUP = 'Test_Resource_Group'\r\nLOCATION = 'australiasoutheast'\r\nSQL_SERVER = 'sourceserver2020'\r\nSQL_DB = 'sourcedb'\r\nUSERNAME = 'testusername'\r\nPASSWORD = 'MyV3ry$trongPa$$word'\r\nDRIVER = '{ODBC Driver 17 for SQL Server}'\r\nexternal_IP = popen("curl -s ifconfig.me").readline()\r\n\r\n\r\ndef create_resource_group(resource_client, RESOURCE_GROUP, LOCATION):\r\n print("\\nCreating Azure RG {rg_name}...".format(\r\n rg_name=RESOURCE_GROUP), end="", flush=True)\r\n try:\r\n resource_client.resource_groups.create_or_update(\r\n RESOURCE_GROUP, {'location': LOCATION})\r\n except CloudError as e:\r\n print(e)\r\n rg = [g.name for g in resource_client.resource_groups.list()]\r\n if RESOURCE_GROUP in rg:\r\n print('OK')\r\n\r\n\r\ndef create_sql_server(sql_client, RESOURCE_GROUP, SQL_SERVER, LOCATION, USERNAME, PASSWORD):\r\n print("Creating Azure SQL Server {ssvr_name}...".format(\r\n ssvr_name=SQL_SERVER), end="", flush=True)\r\n try:\r\n sql_server = sql_client.servers.begin_create_or_update(\r\n RESOURCE_GROUP,\r\n SQL_SERVER,\r\n {\r\n 'location': LOCATION,\r\n 'version': '12.0',\r\n 'administrator_login': USERNAME,\r\n 'administrator_login_password': PASSWORD\r\n }\r\n )\r\n sql_server.wait()\r\n except CloudError as e:\r\n print(e)\r\n ssvr = [i.name for i in sql_client.servers.list_by_resource_group(\r\n RESOURCE_GROUP)]\r\n if SQL_SERVER in ssvr:\r\n print('OK')\r\n\r\n\r\ndef create_sql_db(sql_client, RESOURCE_GROUP, SQL_SERVER, SQL_DB, LOCATION):\r\n print("Creating Azure SQL Database {db_name}...".format(\r\n db_name=SQL_DB), end="", flush=True)\r\n try:\r\n sql_db = sql_client.databases.begin_create_or_update(\r\n RESOURCE_GROUP,\r\n SQL_SERVER,\r\n SQL_DB,\r\n {\r\n 'location': LOCATION,\r\n 'collation': 'SQL_Latin1_General_CP1_CI_AS',\r\n 'create_mode': 'default',\r\n 'requested_service_objective_name': 'Basic'\r\n }\r\n )\r\n sql_db.wait()\r\n except CloudError as e:\r\n print(e)\r\n dbs = [i.name for i in sql_client.databases.list_by_server(\r\n RESOURCE_GROUP, SQL_SERVER)]\r\n if SQL_DB in dbs:\r\n print('OK')\r\n\r\n\r\ndef configure_firewall(sql_client, DRIVER, RESOURCE_GROUP, SQL_SERVER, SQL_DB, USERNAME, PASSWORD, external_IP):\r\n print("Configuring Azure SQL Server Firewall Settings...", end="", flush=True)\r\n try:\r\n sql_client.firewall_rules.create_or_update(\r\n RESOURCE_GROUP,\r\n SQL_SERVER,\r\n "firewall_rule_name_" + external_IP,\r\n {\r\n "startIpAddress": external_IP,\r\n "endIpAddress": external_IP\r\n }\r\n )\r\n except CloudError as e:\r\n print(e)\r\n SQL_SERVER = SQL_SERVER + '.database.windows.net'\r\n with pyodbc.connect('DRIVER='+DRIVER+';SERVER='+SQL_SERVER+';PORT=1433;DATABASE='+SQL_DB+';UID='+USERNAME+';PWD=' + PASSWORD) as conn:\r\n with conn.cursor() as cursor:\r\n cursor.execute("SELECT @@version")\r\n row = cursor.fetchone()\r\n if row:\r\n print('OK')\r\n\r\n\r\ndef main():\r\n # create resource client\r\n resource_client = get_client_from_cli_profile(ResourceManagementClient)\r\n create_resource_group(resource_client, RESOURCE_GROUP, LOCATION)\r\n # create sql client\r\n sql_client = get_client_from_cli_profile(SqlManagementClient)\r\n # create sql server\r\n create_sql_server(sql_client, RESOURCE_GROUP, SQL_SERVER,\r\n LOCATION, USERNAME, PASSWORD)\r\n # create sql db in the basic tier\r\n create_sql_db(sql_client, RESOURCE_GROUP, SQL_SERVER, SQL_DB, LOCATION)\r\n # configure firewall\r\n configure_firewall(sql_client, DRIVER, RESOURCE_GROUP,\r\n SQL_SERVER, SQL_DB, USERNAME, PASSWORD, external_IP)\r\n\r\n\r\nif __name__ == "__main__":\r\n main()\r\n<\/pre>\n![\"\"](\"http:\/\/bicortex.com\/bicortex\/wp-content\/post_content\/\/2020\/12\/Hash_Partitioned_Data_Sync_Azure_DB_Deployment.png\")
<\/a><\/p>\n\r\nfrom pathlib import PureWindowsPath\r\nimport pyodbc\r\n\r\nSQL_SERVER = 'sourceserver2020.database.windows.net'\r\nSQL_DB = 'sourcedb'\r\nUSERNAME = 'testusername'\r\nPASSWORD = 'MyV3ry$trongPa$$word'\r\nDRIVER = '{ODBC Driver 17 for SQL Server}'\r\n\r\n\r\nsql_file = PureWindowsPath(\r\n '\/Path\/Azure_SQLDB_Deployment\/SQL\/create_wide_tbl.sql')\r\n\r\nwith open(sql_file, "r") as f:\r\n sqlFile = f.read()\r\n\r\nsql = sqlFile.rstrip('\\n')\r\n\r\ntry:\r\n with pyodbc.connect('DRIVER='+DRIVER+';SERVER='+SQL_SERVER+';PORT=1433;DATABASE='+SQL_DB+';UID='+USERNAME+';PWD=' + PASSWORD) as conn:\r\n with conn.cursor() as cursor:\r\n cursor.execute('DROP PROCEDURE IF EXISTS usp_generate_dummy_data')\r\n cursor.execute(sql)\r\n cursor.execute('EXEC dbo.usp_generate_dummy_data')\r\n cursor.execute('SELECT TOP (1) 1 FROM dbo.Dummy_Table')\r\n rows = cursor.fetchone()\r\n if rows:\r\n print('All Good!')\r\n else:\r\n raise ValueError(\r\n 'No data generated in the source table. Please troubleshoot!'\r\n )\r\nexcept pyodbc.Error as ex:\r\n sqlstate = ex.args[1]\r\n print(sqlstate)\r\n<\/pre>\n\r\nCREATE PROCEDURE usp_generate_dummy_data\r\nAS\r\nBEGIN\r\n SET NOCOUNT ON;\r\n DECLARE @wide INT = 100;\r\n DECLARE @deep INT = 100000;\r\n DECLARE @allchars VARCHAR(100) = 'ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789 ';\r\n DECLARE @target_table_name VARCHAR(56) = 'Dummy_Table';\r\n\r\n DROP TABLE IF EXISTS ##metadata;\r\n CREATE TABLE ##metadata\r\n (\r\n ID INT IDENTITY(1, 1),\r\n Column_Name VARCHAR(256),\r\n Data_Type VARCHAR(56),\r\n Column_Size VARCHAR(56)\r\n );\r\n\r\n\r\n DECLARE @count INT = 1;\r\n WHILE @count <= @wide\r\n BEGIN\r\n INSERT INTO ##metadata\r\n (\r\n Column_Name,\r\n Data_Type,\r\n Column_Size\r\n )\r\n SELECT 'Column_' + CAST(@count AS VARCHAR(56)),\r\n 'varchar',\r\n CEILING(RAND() * 100);\r\n\r\n SET @count = @count + 1;\r\n END;\r\n\r\n DECLARE @SQL VARCHAR(MAX);\r\n SELECT @SQL\r\n = 'DROP TABLE IF EXISTS ' + @target_table_name + '; CREATE TABLE ' + @target_table_name\r\n + ' (Id INT IDENTITY(1,1),' + STRING_AGG(Column_Name + ' ' + Data_Type + '(' + Column_Size + ')', ',') + ')'\r\n FROM ##metadata;\r\n EXEC (@SQL);\r\n\r\n SET @count = 1;\r\n WHILE @count <= @deep\r\n BEGIN\r\n DECLARE @vals VARCHAR(MAX);\r\n SELECT @vals\r\n = STRING_AGG(\r\n CAST(QUOTENAME(\r\n SUBSTRING(\r\n RIGHT(LEFT(@allchars, ABS(BINARY_CHECKSUM(NEWID()) % 35) + 5), 5)\r\n + RIGHT(LEFT(@allchars, ABS(BINARY_CHECKSUM(NEWID()) % 35)\r\n + 5), 5)\r\n + RIGHT(LEFT(@allchars, ABS(BINARY_CHECKSUM(NEWID()) % 35)\r\n + 5), 5)\r\n + RIGHT(LEFT(@allchars, ABS(BINARY_CHECKSUM(NEWID()) % 35)\r\n + 5), 5)\r\n + RIGHT(LEFT(@allchars, ABS(BINARY_CHECKSUM(NEWID()) % 35)\r\n + 5), 5)\r\n + RIGHT(LEFT(@allchars, ABS(BINARY_CHECKSUM(NEWID()) % 35)\r\n + 5), 5)\r\n + RIGHT(LEFT(@allchars, ABS(BINARY_CHECKSUM(NEWID()) % 35)\r\n + 5), 5)\r\n + RIGHT(LEFT(@allchars, ABS(BINARY_CHECKSUM(NEWID()) % 35)\r\n + 5), 5)\r\n + RIGHT(LEFT(@allchars, ABS(BINARY_CHECKSUM(NEWID()) % 35)\r\n + 5), 5)\r\n + RIGHT(LEFT(@allchars, ABS(BINARY_CHECKSUM(NEWID()) % 35)\r\n + 5), 5),\r\n 1,\r\n CAST(Column_Size AS INT)\r\n ),\r\n ''''\r\n ) AS NVARCHAR(MAX)),\r\n ','\r\n )\r\n FROM ##metadata;\r\n SELECT @SQL\r\n = 'INSERT INTO ' + @target_table_name + '(' + STRING_AGG(Column_Name, ',') + ') SELECT ' + @vals + ''\r\n FROM ##metadata;\r\n EXEC (@SQL);\r\n SET @count = @count + 1;\r\n END;\r\n\r\n ALTER TABLE dbo.Dummy_Table\r\n ADD PRIMARY KEY (ID);\r\n\r\n DROP TABLE IF EXISTS ##metadata;\r\nEND;\r\n<\/pre>\n![\"\"](\"http:\/\/bicortex.com\/bicortex\/wp-content\/post_content\/\/2020\/12\/Hash_Partitioned_Data_Sync_Dummy_Data.png\")
<\/a><\/p>\n\r\nUSE [master]\r\nGO\r\nEXEC master.dbo.sp_addlinkedserver @server = N'AZURESOURCELINKEDSVR', @srvproduct=N'', @provider=N'SQLNCLI', @datasrc=N'sourceserver2020.database.windows.net', @catalog=N'sourcedb'\r\nGO\r\nEXEC master.dbo.sp_addlinkedsrvlogin @rmtsrvname=N'AZURESOURCELINKEDSVR',@useself=N'False',@locallogin=NULL,@rmtuser=N'testusername',@rmtpassword='MyV3ry$trongPa$$word'\r\nGO\r\nEXEC master.dbo.sp_serveroption @server=N'AZURESOURCELINKEDSVR', @optname=N'collation compatible', @optvalue=N'false'\r\nGO\r\nEXEC master.dbo.sp_serveroption @server=N'AZURESOURCELINKEDSVR', @optname=N'data access', @optvalue=N'true'\r\nGO\r\nEXEC master.dbo.sp_serveroption @server=N'AZURESOURCELINKEDSVR', @optname=N'dist', @optvalue=N'false'\r\nGO\r\nEXEC master.dbo.sp_serveroption @server=N'AZURESOURCELINKEDSVR', @optname=N'pub', @optvalue=N'false'\r\nGO\r\nEXEC master.dbo.sp_serveroption @server=N'AZURESOURCELINKEDSVR', @optname=N'rpc', @optvalue=N'true'\r\nGO\r\nEXEC master.dbo.sp_serveroption @server=N'AZURESOURCELINKEDSVR', @optname=N'rpc out', @optvalue=N'true'\r\nGO\r\nEXEC master.dbo.sp_serveroption @server=N'AZURESOURCELINKEDSVR', @optname=N'sub', @optvalue=N'false'\r\nGO\r\nEXEC master.dbo.sp_serveroption @server=N'AZURESOURCELINKEDSVR', @optname=N'connect timeout', @optvalue=N'0'\r\nGO\r\nEXEC master.dbo.sp_serveroption @server=N'AZURESOURCELINKEDSVR', @optname=N'collation name', @optvalue=NULL\r\nGO\r\nEXEC master.dbo.sp_serveroption @server=N'AZURESOURCELINKEDSVR', @optname=N'lazy schema validation', @optvalue=N'false'\r\nGO\r\nEXEC master.dbo.sp_serveroption @server=N'AZURESOURCELINKEDSVR', @optname=N'query timeout', @optvalue=N'0'\r\nGO\r\nEXEC master.dbo.sp_serveroption @server=N'AZURESOURCELINKEDSVR', @optname=N'use remote collation', @optvalue=N'true'\r\nGO\r\nEXEC master.dbo.sp_serveroption @server=N'AZURESOURCELINKEDSVR', @optname=N'remote proc transaction promotion', @optvalue=N'false'\r\nGO\r\n<\/pre>\n
![\"\"](\"http:\/\/bicortex.com\/bicortex\/wp-content\/post_content\/\/2020\/12\/Hash_Partitioned_Data_Sync_Metadata_Schema.png\")
<\/a><\/p>\n