# Snowflake

### Before You Begin

Access to your Snowflake Console is required to complete this integration guide.

### Configure a Snowflake Storage Integration and Stored Procedure

**Step 1**

Create a storage integration using the `CREATE STORAGE INTEGRATION COMMAND`

```sql
CREATE STORAGE INTEGRATION clario_integration
type = external_stage
storage_provider = gcs
enabled = true
storage_allowed_locations = ('gcs://<bucket>');
```

Retrieve the Cloud Storage Service Account for your Snowflake Account. Provide the value of `STORAGE_GCP_SERVICE_ACCOUNT` to your Clario Customer Success team.

```sql
DESC STORAGE INTEGRATION clario_integration;
```

**Step 2**

Create a stored procedure to handle the export:

```sql
create or replace procedure clario_unloader_sp(brand varchar, sourceType varchar, feedType varchar, db varchar, dbschema varchar, dbtable varchar)
returns string not null
language javascript
as
$$
var today = new Date().toISOString().slice(0,10).replace('-','').replace('-','');
var sql_command = 'copy into '.concat(
"'gcs://<bucket>/",FEEDTYPE,"/",BRAND,"/",SOURCETYPE,"/",today,"/'",
' from "', DB,'"."', DBSCHEMA, '"."', DBTABLE, '"',
" file_format = (type='PARQUET', compression='SNAPPY')",
" header = true",
" include_query_id = true",
" storage_integration = clario_integration");
try {
var statement = snowflake.createStatement({sqlText: sql_command});
statement.execute();
return sql_command;
} catch (err) {
return "failed " + err;
}
$$;
```

### Schedule Unload Jobs

For each table you wish to unload, create a task using a [schedule](https://docs.snowflake.com/en/sql-reference/sql/create-task.html#optional-parameters) that fits your needs.

```sql
CREATE TASK clario_unload_<brand>_<sourceType>_<feedType>
warehouse = <your warehouse name>
schedule = 'USING CRON <cron definition>'
AS
CALL clario_unloader_sp('<brand>', '<sourceType>', '<feedType>', '<db>', '<schema>', '<table>');
```

The inputs for `clario_unloader_sp` are as follows:

<table data-header-hidden><thead><tr><th width="189">Input</th><th>Description</th></tr></thead><tbody><tr><td><strong>brand</strong></td><td>Should be a meaningful abbreviation of the brand you are sending data for (i.e. Avalanche Clothing would be ac).</td></tr><tr><td><strong>sourceType</strong></td><td>Should represent the type of data you are sending (i.e. customer_master, product_master, esp_activity, esp_subsciber, etc.).</td></tr><tr><td><strong>feedType</strong></td><td><em>ongoing</em> (for recurring feeds) or <em>backfill</em> (for one-time historical loads).</td></tr><tr><td><strong>db</strong></td><td>The name of the Snowflake database your table lives in.</td></tr><tr><td><strong>schema</strong></td><td>The schema your Snowflake table is organized under.</td></tr><tr><td><strong>table</strong></td><td>The Snowflake table you are unloading to GCS.</td></tr></tbody></table>

{% hint style="info" %}
Note:

All unload tasks are suspended upon creation. Ensure you enable them.

`ALTER TASK IF EXISTS clario_unload_<brand>_<sourceType>_<feedType> RESUME;`
{% endhint %}

### Example Unload Job

You work on the data team for Avalanche Clothing. You are in the process of onboarding with Clario and need to get a feed established to pull transaction header data out of your Snowflake warehouse. Clario is asking for a historical backfill and an ongoing feed. The following assumptions can me made:

* Warehouse: `AVALANCHE_WH`
* Brand: `ac` (**A**valanche **C**lothing)
* Source Type: `transaction_header`
* Schema: `ERP`
* Table: `TR_HEADER`
* Scheduled tasks will run at 3:00 AM daily, UTC
* The data team maintains two databases in the warehouse: `HISTORICAL` and `CURRENT`.

### Schedule and Enable the Backfill <a href="#xni8x12d2huu" id="xni8x12d2huu"></a>

```sql
CREATE TASK clario_unload_ac_transaction_header_backfill
warehouse = AVALANCHE_WH
schedule = 'USING CRON 0 3 * * * UTC'
AS
CALL clario_unloader_sp('ac', 'transaction_header', 'backfill', 'HISTORICAL', 'ERP', 'TR_HEADER');
ALTER TASK IF EXISTS clario_unload_ac_transaction_header_backfill RESUME;
```

### Schedule and Enable the Ongoing Feed <a href="#cp0cb2luqxe3" id="cp0cb2luqxe3"></a>

```sql
CREATE TASK clario_unload_ac_transaction_header_ongoing
warehouse = AVALANCHE_WH
schedule = 'USING CRON 0 3 * * * UTC'
AS
CALL clario_unloader_sp('ac', 'transaction_header', ongoing, 'CURRENT, 'ERP', 'TR_HEADER');
ALTER TASK IF EXISTS clario_unload_ac_transaction_header_ongoing RESUME;
```

### Working With Scheduled Jobs

### List <a href="#vwmnicpqkzkm" id="vwmnicpqkzkm"></a>

```sql
SHOW TASKS;
Delete
DROP TASK IF EXISTS <task name>;
```

### View History <a href="#vacmzbyom5q0" id="vacmzbyom5q0"></a>

```sql
select *
from table(information_schema.task_history())
order by scheduled_time DESC;
```


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.clar.io/connecting-your-data/snowflake.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.