This sets up a Kafka connect cluster with a Snowflake connector. 
This POC uses `dev-managed-kafka-neo4j-streams` MSK brokers to connect to the `default_node_topic`. 
It's configured to sink the data to `DEV_ENGINEERING.TEST_SCHEMA`.

Snowflake ssh key authentication setup.
```
Create a private key. You will be prompt for a paraphrase. This key is (FIPS 140-2) compliant.
openssl genrsa 2048 | openssl pkcs8 -topk8 -v2 aes256 -inform PEM -out private.p8

Create public key.
openssl rsa -in private.p8 -pubout -out public.pub

Upload the public key to snowflake.

To copy your ssh private key cleanly to clipboard without a trailing newline use
cat private.p8 | tr -d '\n' | pbcopy

Paste the key in .env file SNOWFLAKE_PRIVATE_KEY=paste_here
```

The .env file to be mounted requires the following items:
```
SNOWFLAKE_USER=myuser
SNOWFLAKE_PRIVATE_KEY= MIIE6TAbBgkqhkiG9w0BB.............
SNOWFLAKE_HOST=orchard.snowflakecomputing.com:443
SNOWFLAKE_PRIVATE_KEY_PASSPHRASE=mypassphrase
```

After this setup, kafka events will start showing up in the designated destination topic table in Snowflake:
```
SELECT * FROM DEV_ENGINEERING.TEST_SCHEMA.DEFAULT_NODE_TOPIC LIMIT 5;
```

If working with event data like neo4j cdc, these will be the exact transaction logs from the database.
In order to transform them back into structured tables we will be using Snowflake's STREAMS and TASKS.

First, create the destination table:
```
CREATE OR REPLACE table DEV_ENGINEERING.TEST_SCHEMA.neo4j_label_participant (uuid string, name string, vendor_id int);
```

Then, create a stream on the transaction log table:
```
CREATE OR REPLACE stream default_node_topic_stream on table "DEV_ENGINEERING"."TEST_SCHEMA"."DEFAULT_NODE_TOPIC" ;
```

Lastly, create a scheduled task that's hooked up to any changes on the newly created stream:
```
create or replace task cdc_to_participant
warehouse = DEV_OWS_WAREHOUSE
schedule = '1 minute'
when
system$stream_has_data('default_node_topic_stream')
as
merge into DEV_ENGINEERING.TEST_SCHEMA.neo4j_label_participant nlp
  using (SELECT RECORD_CONTENT:payload:after:properties:uuid::string as uuid,
        RECORD_CONTENT:payload:after:properties:name::string as name,
        RECORD_CONTENT:payload:after:properties:vendorId::int as vendor_id
        FROM default_node_topic_stream
        WHERE RECORD_CONTENT:payload:after:properties:uuid IS NOT NULL
        ) raw on nlp.uuid = raw.uuid
  when matched then update set nlp.name = raw.name, nlp.vendor_id = raw.vendor_id
  when not matched then insert (uuid, name, vendor_id) values (raw.uuid, raw.name, raw.vendor_id);
```

The task always gets created in suspend mode, so let's resume it:
```
alter task cdc_to_participant resume;
```

The new table will start populating every minute with changes that are tracked by the stream.
Here's a handy query that can provide more information about the scheduled tasks:
```
select *
  from table(information_schema.task_history())
  order by scheduled_time;
```


Reference material can be found here:
https://docs.snowflake.com/en/user-guide/kafka-connector-overview.html
https://rmoff.net/2019/11/20/streaming-data-from-sql-server-to-kafka-to-snowflake-with-kafka-connect/
https://github.com/confluentinc/demo-scene/blob/master/pipeline-to-the-cloud/docker-compose_auto-provision.yml