Set up DataSync Agent to share to Snowflake

You can configure your DataSync Agent to share data from ServiceNow to a Snowflake database by changing some additional configurations in your agent.xml file. 

The DataSync Agent uses the Snowflake COPY command to bulk load data from the local file system where the Agent is running to optimize performance. Being that Snowflake is a cloud-based database, the COPY command is used to improve performance as standard SQL operations (i.e. INSERT, UPDATE, etc) on a record by record basis will have slow performance due to network latency.

Prerequisites

 First, you will need to Install the DataSync Agent.

 You will also need to create a ServiceNow dynamic share/bulk share.

 Make sure to stop running your DataSync Agent before making any Agent configuration changes.

Procedure

To set up your DataSync Agent to share application data to Snowflake, follow these steps:

<?xml version="1.0" encoding="ISO-8859-1" standalone="no"?>
<config>
    <agent>
       	<share>
     	</share>  
     	<subscribe>
     	<task>
            <task_name>test_snowflake_subscribe</task_name>
            <message_connection password="encrypted:vlOtU71yu8N/EFIJH85SSBtaIt7qEEfvqiqft9VZyYE=" user="exampleuser">amqps://example.perspectium.net</message_connection>
            <instance_connection password="encrypted:vlOtU71yu8N/EFIJH85SSBtaIt7qEEfvqiqft9VZyYE=" user="example.user">https://myinstance.service-now.com</instance_connection>     
			<decryption_key>The cow jumped over the moon</decryption_key>
            
			<handler>com.perspectium.replicator.sql.SnowflakeSQLSubscriber</handler>             
			<database_type>snowflake</database_type>
            <database_server>servername.snowflakecomputing.com</database_server>
            <database_user>exampleuser</database_user>
            <database_password>examplepassword</database_password>
            <database_params>schema=PUBLIC&warehouse=COMPUTE_WH&role=ACCOUNTADMIN</database_params>
            <database>SAMPLE</database>
			
			<enable_snowflake/>
			<uppercase_columns/>
				
			<files_directory>/usr/local/datasyncagent/tempfiles</files_directory>
			<file_max_size>500KB</file_max_size>
			<flush_interval>3</flush_interval>				
			<delete_files_interval>12H</delete_files_interval>          
		</task>
      	</subscribe>
      	<max_reads_per_connect>4000</max_reads_per_connect>
      	<polling_interval>30</polling_interval>
   </agent>
</config>

wrapper.java.additional.4=--add-opens 
wrapper.java.additional.5=java.base/java.nio=ALL-UNNAMED

# Java Additional Parameters
wrapper.java.additional.1=-Djava.util.logging.config.file=../conf/logging.properties
wrapper.java.additional.2=-Dlog4j.configurationFile=../conf/log4j2.xml
wrapper.java.additional.3=-Dfile.encoding=UTF-8
wrapper.java.additional.4=--add-opens 
wrapper.java.additional.5=java.base/java.nio=ALL-UNNAMED  

Notable Features

Agent Log Reporting

The Agent reports both the records it processes (i.e. records it reads from the Perspectium Integration Mesh queue and then prepares for saving into Snowflake) along with the records it pushes into Snowflake (records that are sent into Snowflake and then copied into the database tables with the COPY command). Because of how the Agent batches records together to a max file size before using the COPY command, these details will be reported separately and these two actions will not happen at the exact same time. For example, if looking at the Agent's log file (perspectium.log), you will see log entries as follows:

2024-03-28 23:33:51.049 INFO  - PerformanceReportGenerator - PerformanceReport - change_request.bulk=[100, 425560]: 100 messages processed and 425560 bytes processed

2024-03-28 23:37:21.049 INFO  - Timer-4 - SnowflakeSQLSubscriber - Scheduled flushing to push records

2024-03-28 23:37:28.144 INFO  - Timer-4 - SnowflakeSQLSubscriber - Processed 100 record(s) to table: change_request

The PerformanceReportGenerator logging will represent the processing of records by the Agent from the Integration Mesh queue (i.e. reading the messages, decrypting the content, formatting it and then saving into the local temporary files) and then the SnowflakeSQLSubscriber logging will represent the actual pushing of the records into the Snowflake database (taking the temporary files, pushing them into Snowflake and then using the COPY command to save into the actual Snowflake database tables themselves).

Column Sizes

By default, the DataSync Agent will create columns to match the sizes as specified in ServiceNow up to 4000 characters. Anything that is defined over 4000 characters will be created as the maximum size for that column type.

For example, if in ServiceNow the column is defined as String with 200 characters, in Snowflake this column will be created as VARCHAR(200) (VARCHAR is the equivalent for a String in the Snowflake database).

If the column was defined as String with 10,000 characters in ServiceNow, in Snowflake this column will be created as VARCHAR(16777216) which is the maximum size for VARCHAR column types since Snowflake does not have a TEXT or CLOB equivalent.

You can update the default value of 4000 that controls when the column is created as the maximum size by updating the <column_max_size> configuration in databases.xml. This value can only be changed to the maximum size that Snowflake supports (i.e. you can't set this value to 32777216 since this is greater than the VARCHAR maximum value of 16777216).

Database Views

A ServiceNow database view replicated to the Agent for saving into Snowflake will use the encoded sys_id value so that the record can be properly queried both for updating and deleting. This sys_id value is created by ServiceNow when the Perspectium application queries for records from a database view for sharing. The sys_id will have a value such as __ENC__MzVmM2RjNjljMGE4MDhhZTAwMDg2NjU3MzFjMTdkMDM=-MjQ2ZWY2ZDk4MzE5NDIxMGYyZGQxNjMwY2VhYWQzZWE=-MDg2ZTM2ZDk4MzE5NDIxMGYyZGQxNjMwY2VhYWQzYmQ=.

 NOTE: The legacy DataSync Agent approach with combining sys_ids based on *_sys_id fields listed in the record for database views is not supported.

Table Compare

In order for Table Compare to work properly, the databases.xml configuration file needs two configurations (<most_recent_record_query> and <all_ids_query_syntax>) to properly query for records in tables being shared to Snowflake. These configurations should be located under the <database> section where <database_type>snowflake</database_type> is (note other configurations have been omitted in the example below for easy viewing):

 			<database>
                <database_type>snowflake</database_type>
 				.
				.
				.
                <most_recent_record_query>SELECT sys_id FROM %s WHERE sys_id = '%s' AND %s = %s</most_recent_record_query>
                <all_ids_query_syntax>SELECT SYS_ID as "sys_id" FROM %s</all_ids_query_syntax>
            </database>

Because of how the Agent queries for records when synchronizing deletes, the <all_ids_query_syntax> should contain the "sys_id" alias for the sys_id field. The default configuration uses SYS_ID as this field and aliases it to "sys_id" (i.e. SYS_ID as "sys_id") since SYS_ID in all upper case lettering is Snowflake's default syntax for column names. You should update this if you have different casing for your column names.

Contact Perspectium Support to obtain the latest version and after verifying the two configurations are in databases.xml, place it in the Agent's conf folder i.e. <Perspectium_Replicator_Agent_Installed_Directory>/conf.  

agent.ack and agent.error Messages

Since we are leveraging the COPY API, the Agent's processing of messages it receives and pushing the data to Snowflake are two separate processes (whereas with other databases, the Agent will process a message and push it to the database in one step). As a result, the Agent will send back agent.ack messages for processing messages (not including pushing data to Snowflake) and then if the Agent has any errors pushing this data to Snowflake, will send back agent.error messages. This way you can subscribe to notifications for these error messages.

IODatetime Deletes

For deletes with the IODatetime plugin, the column specified as the update_column will also be updated with the datetime when the delete action happens. This is due to how the COPY API updates the records to mark them as a soft delete (i.e. not actually deleted from the database table).