# Snowflake Sink Connector

## Description

The Snowflake Kafka connector reads data from one or more Apache Kafka topics and loads the data into a Snowflake table. In general, each Kafka message contains one row.

### Snowflake access configuration

You have to provide following environment variables. If they are not in env then the connector will look for them in secretsmanager.
- SNOWFLAKE_HOST - The URL for accessing your Snowflake account. This URL must include your account identifier. Note that the protocol (https://) and port number are optional.
- SNOWFLAKE_USER - User login name for the Snowflake account.
- SNOWFLAKE_PRIVATE_KEY - The private key to authenticate the user. Include only the key, not the header or footer. If the key is split across multiple lines, remove the line breaks.
- SNOWFLAKE_PRIVATE_KEY_PASSPHRASE - phrase to try to decrypt the private key.


For local testing create and fill up `.env` file and run 
```
awsume dev
aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin 086679231553.dkr.ecr.us-east-1.amazonaws.com
docker-compose up --build -d
```

### Kafka topics to Snowflake table mapping

All topics are specified as comma seperted list in env: `KAFKA_TOPICS`. Each topic must also have a corresponding table mapped in env `SNOWFLAKE_TOPIC_TABLE_MAP`. Eg: `cdc.musicGraph.globalSoundRecording:CDC_MUSICGRAPH_GLOBALSOUNDRECORDING`. If the topics are not mapped, then the Kafka connector creates a new table for each topic using the topic name. More details on naming can be found [in their Official documentation](https://docs.snowflake.com/en/user-guide/kafka-connector-overview#target-tables-for-kafka-topics).

All corresponding tables will be stored in a single database and schema specified by env: `SNOWFLAKE_DATABASE` and `SNOWFLAKE_SCHEMA`


### Schema of target Snowflake tables

Snowflake table loaded by the Kafka connector has a schema consisting of two VARIANT columns:
- RECORD_CONTENT - This contains the Kafka message. The internal structure of the message is JSON in our case.
- RECORD_METADATA - This contains metadata about the message, for example, the topic from which the message was read.

The Kafka connector buffers messages from the Kafka topics. When a threshold (time or memory or number of messages) is reached, the connector writes the messages to a temporary file in the internal stage. You can configure these via `buffer.*` configurations. More details on the message schema can be found [in their Official documentation](https://docs.snowflake.com/en/user-guide/kafka-connector-overview#schema-of-tables-for-kafka-topics).


**Note:** When running the connector for the first time and a new snowflake table is created, it wont be immediately visible in snowflake. This is because our users dont have proper permissions granted to new tables. This [scheduler job](https://scheduler.theorchard.io/job/snowflake-db-refresh/build?delay=0sec) runs every day to fix that but in case you want to check immediately you can explicitly run the job. It is connected to run [this util](https://github.com/theorchard/sql-snowflake-utils/blob/master/TEMP_facts_table_refresh.sql#L319-L320) script.


### Snowflake sink with Snowpipe Streaming
Starting with version 2.0.0 of the Snowflake Kafka Connector, Snowflake introduced support for the Snowpipe Streaming API. 
This update enables near real-time data ingestion. Read more about it in [Notion documentation](https://www.notion.so/Snowflake-sink-Streaming-API-versus-Snowpipe-1de97177520f800496e2dddadd680535) and their [official documentation](https://docs.snowflake.com/en/user-guide/data-load-snowpipe-streaming-overview). 
Also 
- Snowflake Key and Value converters are not supported with Snowpipe Streaming. So the conenctor will ignore `CONNECT_KEY_CONVERTER and CONNECT_VALUE_CONVERTER` environment variable settings. 
- TRANSIENT or TEMPORARY tables are not supported. So if you are switching an existing sink connector over to this then check if your CDC tables are regular and not transient.


You have to provide following environment variables:
```
SNOWFLAKE_INGESTION_METHOD="SNOWPIPE_STREAMING"
SNOWFLAKE_ROLE="some snowflake role eg DEV_ENGINEERING"

Optional configurations: 
SNOWFLAKE_ENABLE_SINGLE_BUFFER="true or false to  skip buffering data in the connector's internal buffer. Default is true."

```


### Sample terraform setup
https://github.com/theorchard/terraform-infra/blob/master/prod/kafka-infra/snowflake_sink/main.tf#L46


### Sample SQL to view the Snowflake table

```
SELECT
    RECORD_CONTENT:payload:id::number as rel_id,
    RECORD_CONTENT:payload:end:ids:id::string as agreement_id,
    RECORD_CONTENT:payload:start:ids:id::string as composition_id,
    RECORD_CONTENT:payload:after:properties:lastModifiedBy::STRING AS last_modified_by,
    RECORD_CONTENT:meta:timestamp as ts,  -- changed to match task defination
    RECORD_CONTENT:meta:operation::string as operation,
    RECORD_CONTENT:meta:txId::number as txid,
    RECORD_CONTENT:meta:txEventId::number as tx_event_id
FROM FACTS.PROD.CDC_MUSICGRAPH_HASAGREEMENT
limit 10;

```