Mule Snowflake Operations With Snowflake Configuration

This tutorial covers a few operations in the Mule Snowflake Connector along with the Snowflake configuration for those operations.

Sajal Biswas

Oct. 05, 22 · Tutorial

Likes (1)

Comment

Save

5.5K Views

In this post, we will cover a few operations in the Mule Snowflake Connector along with the Snowflake configuration for those operations (task, store procedure, multi-table insert, merge, row listener, and Snowpipe concept).

Snowflake Access and Setup

Snowflake Access for Development

Sign-up process:
1. Please create an account in Snowflake. Please note:
  - Account name: We need to mention from the URL we can infer SB15234.ap-southeast-1.
  - For the trial account, you would have 30 days. This allows you access to the same database with data. You can explore for your test and development.
2. Select cloud hosting (Standard) and region (Asia Pacific), Cloud Provider AWS (out of any AWS, Google, Azure) would be fine. For the rest, enter the information asked based on which purpose you need.
3. You will receive an activation mail along with a Snowflake-hosted URL, which will be used for the Snowflake login from the browser. For the first time, you need to set the user name and password for the Snowflake account.
Enterprise work process: Generally enterprise applications or licensed applications would either allow single sign-on or they will create a service account for Snowflake access.

3. You can connect as well using Snowflake CLI.

Snowflake Security

The Snowflake side admin team can raise the IP whitelisting policy. If the client VPN ranges, we know then the entire CIDR block can be whitelisted. This means Snowflake can be accessed only from the whitelisted IP.
We can create users and have specific roles. Roles are comprised of different permission sets. Users who have relevant roles can grant permission to other users with roles as well.

Mule Snowflake Connector Configuration

In the MuleSoft code below, I have entered a dummy account name, username, and password. Please create your own account and fill in those details. (Note: you can use a database generic connector.)

XML

<snowflake:snowflake-config name="Snowflake_Config" doc:name="Snowflake Config" doc:id="45b2e5a4-0d03-4d65-8490-e528191b7dd5" >
<snowflake:snowflake-connection accountName="SB15234.ap-southeast-1" warehouse="COMPUTE_WH" database="snowflake_sample_data" schema="weather" user="*****" password="*****" role="ACCOUNTADMIN" ></snowflake:snowflake>
</snowflake:snowflake-config>

<db:config name="Database_Config" doc:name="Database Config" doc:id="f37b9c9f-4a1d-44c0-a299-62e759bbf7d5" >
<db:generic-connection url="jdbc:snowflake://SB15234.ap-southeast-1.snowflakecomputing.com/?user=dev30&password=dev30" driverClassName="net.snowflake.client.jdbc.SnowflakeDriver" user="dev30" password="dev30" ></db:generic>
</db:config>

Snowflake Worksheet Select Query

Snowflake worksheet: If you select database and schema, then in the query you only need to mention the table name.

Mule Snowflake Select Query

XML

<snowflake:select doc:name="Select" doc:id="1280fe1e-e079-41d6-9f50-b2b2dc191943" config-ref="Snowflake_Config">
<snowflake:sql ><![CDATA[select * from snowflake_sample_data.weather.daily_14_total limit 10]]></snowflake:sql>
</snowflake:select>

<db:select doc:name="Select" doc:id="ac382ae8-e887-4d79-828d-7a1abcf64a60" config-ref="Database_Config">
<db:sql ><![CDATA[select * from snowflake_sample_data.weather.daily_14_total limit 10]]></db:sql>
</db:select>

Snowflake Store Procedure

From the Snow sight view, go to Database -> schema-> create. If you click then you will get these options in the dropdown. Let’s talk about store procedure example:

SQL

CREATE OR REPLACE PROCEDURE POCADMINDB.PUBLIC.GETEMPLOYEE()
returns table(id varchar,name varchar,salary number)
language sql
EXECUTE AS OWNER
AS '
-- Snowflake Scripting code
declare
res resultset default (select * from "POCADMINDB"."PUBLIC"."EMPLOYEES");
begin
return table(res);
end;
';

Mule Snowflake Calling Store Procedure

Note: Please make sure which database and schema the store procedure is written. We need to use that Snowflake Connector configuration to refer to the proper location of Snowflake.

XML

<snowflake:stored-procedure doc:name="Stored procedure" config-ref="Snowflake_Config">
<snowflake:sql ><![CDATA[ call GETEMPLOYEE();]]></snowflake:sql>
</snowflake:stored-procedure>

Snowflake Query Scheduling Task

You can create a task in Snowflake under any schema. Under the task, you can define SQL and call a store procedure as well. You can also give cron expression (ex: SCHEDULE = 'USING CRON 0 0 10-20 * TUE,THU UTC' ) while defining tasks. Let's have a look.

SQL

create task insert_data_interval
warehouse = 'COMPUTE_WH'
schedule = '1 MINUTE'
as
call myprocedure();

After defining you can start or stop using the below commands.

SQL

alter task insert_data_interval resume;
alter task insert_data_interval suspend;

Mule Snowflake Creating Task Query

This will create or override the scheduler programmatically from Mule. Please note that timeIntervalInMinutes and cronExpression can’t be used at the same time.

XML

<snowflake:create-task doc:name="Create task" doc:id="3094dfc4-159a-419b-9ea6-ccbdef2f37a4" config-ref="Snowflake_Config" taskName="compute_data_interval" warehouseName="COMPUTE_WH" sql="select * from POCADMINDB.PUBLIC.EMPLOYEES" replaceExistingTask="true">

<snowflake:schedule-option-cron-expression cronExpression='5 20 * * *' timeZone='UTC' />
</snowflake:create-task>

Snowflake Multiple Table Insert

We can select data from a particular table and insert that data into multiple tables based on conditions. For the below example, salary is a field from the employees' table. We put some conditions based on that it will insert data to two different tables.

SQL

insert all
when salary > 10000 then
into "POCADMINDB"."PUBLIC".MANAGER
else
into "POCADMINDB"."PUBLIC"."STAFF"
select * from "POCADMINDB"."PUBLIC"."EMPLOYEES";

MuleSoft Snowflake Multiple Table Insert

The above example can be written from MuleSoft code as below.

XML

<snowflake:insert-multi-table doc:name="Insert multi table" config-ref="Snowflake_Config">
<snowflake:insert-into-multi-table-parameters >
<snowflake:conditional executionMode="ALL" subquery='select * from POCADMINDB.PUBLIC.EMPLOYEES' elseIntoClause="#['into STAFF']">
<snowflake:conditions >
<snowflake:condition-statement whenCondition="salary > 10000">
<snowflake:into-clauses >
<snowflake:into-clause value="#['into MANAGER']" />
</snowflake:into-clauses>
</snowflake:condition-statement>
</snowflake:conditions>
</snowflake:conditional>
</snowflake:insert-into-multi-table-parameters>
</snowflake:insert-multi-table>

Snowflake SQL Merge

SQL

merge into STAFF
using EMPLOYEES on STAFF.salary = EMPLOYEES.salary
when matched then update
set STAFF.salary = EMPLOYEES.salary
when not matched then
insert (ID, NAME,SALARY) values (EMPLOYEES.ID , EMPLOYEES.NAME , EMPLOYEES.SALARY);

MuleSoft Snowflake Merge

The below example will merge and update the staff table from the employees' table where salaries are equal. This will produce the same result as the above query.

XML

<snowflake:merge doc:name="Merge" doc:id="1a8986fa-1cd1-444d-837e-a78d1d3ab3f8" config-ref="Snowflake_Config" targetTable="staff" source="EMPLOYEES" joinExpr="EMPLOYEES.SALARY = staff.salary">
<snowflake:match-clauses >
<snowflake:not-matched-insert-clause>
<snowflake:column-names >
<snowflake:column-name value="#['ID']" />
<snowflake:column-name value="#['NAME']" />
<snowflake:column-name value="#['SALARY']" />
</snowflake:column-names>
<snowflake:column-values >
<snowflake:column-value value="EMPLOYEES.ID" />
<snowflake:column-value value='EMPLOYEES.NAME' />
<snowflake:column-value value="EMPLOYEES.SALARY" />
</snowflake:column-values>
</snowflake:not-matched-insert-clause>
</snowflake:match-clauses>
</snowflake:merge>

Mulesoft Snowflake Ontable Row Listener

In the example below, if any new record is inserted with more than the previous ID value, the Mule event will then be automatically triggered.

XML

<snowflake:row-listener table="STAFF" doc:name="On Table Row" doc:id="05eea0b2-5bb2-431e-9f11-3d265191de9e" config-ref="Snowflake_Config" watermarkColumn="ID" idColumn="ID">
<scheduling-strategy >
<fixed-frequency frequency="3000" />
</scheduling-strategy>
</snowflake:row-listener>

Snowflake Snowpipe and Staging Concept

We can consume data from AWS, Google, and Azure cloud using copy command. You can check more details from the Snowflake documentation's Introduction to Snowpipe. Staging is where we used to store data from cloud places. After that, we can automate this using series of SQL transformations for loading into different tables.

MULE

Opinions expressed by DZone contributors are their own.

Related

Trending