Completed pipeline

Author: rmoff

README.md

@@ -35,10 +35,11 @@ _Learn more [here](https://decodable.co), and [sign up for a free trial](https:/
 
 ### Data Pipelines
 
-| Example                                             | Description                                            |
-|-----------------------------------------------------|--------------------------------------------------------|
-| [Opinionated Data Pipelines](opinionated-pipelines) | Building data pipelines with schema on write streams. |
-| [Postman](postman)                                  | Building data pipelines with Postman.                  |
+| Example                                             | Description                                              |
+|-----------------------------------------------------|----------------------------------------------------------|
+| [Opinionated Data Pipelines](opinionated-pipelines) | Building data pipelines with schema on write streams.    |
+| [Postman](postman)                                  | Building data pipelines with Postman.                    |
+| [Postgres to Snowflake](postgres-to-snowflake-with-cdc)                    | Getting data from Postgres to Snowflake using Decodable. |
 
 ### PyFlink
 

postgres-to-snowflake-with-cdc/README.adoc

@@ -1,4 +1,8 @@
-= Docker Compose for running Postgres locally, accessible from the internet using ngrok
+= Postgres CDC to Snowflake
+
+This is supporting code for a blog about streaming data from Postgres to Snowflake using Decodable.
+
+It includes a Docker Compose for running Postgres locally, accessible from the internet using ngrok
 
 To use this you need to https://dashboard.ngrok.com/signup[create an ngrok account] and add a file called `.env` in this folder with the following entry:
 
@@ -14,14 +18,14 @@ Bring up the Postgres and ngrok stack with
 docker compose up
 ----
 
-Once up, find out your Postgres server host/post that is available on the internet:
+Once up, you can find out your Postgres server host/post that is available on the internet:
 
 [source,bash]
 ----
 curl -s localhost:4040/api/tunnels | jq -r '.tunnels[0].public_url' | sed 's/tcp:\/\///g'
 ----
 
-== Steps
+== Source connection: Postgres CDC
 
 === Configure Postgres tables for replication
 
@@ -41,38 +45,16 @@ Check their replica status; each should show `f`:
 
 [source,sql]
 ----
-SELECT relreplident FROM pg_class
+SELECT oid::regclass, relreplident FROM pg_class
  WHERE oid in ( 'customers'::regclass, 'pets'::regclass, 'appointments'::regclass);
 ----
 
-=== Add PKs to the Postgres tables
-
-[source,bash]
-----
-docker exec -it postgres psql -h localhost -U postgres -d postgres -f /data/add-pk.sql
-----
-
-
-[source,sql]
-----
-alter table customers
-alter column customer_id set not null;
-alter table customers
-add constraint pk_customers primary key (customer_id);
-
-alter table pets
-alter column pet_id set not null;
-alter table pets
-add constraint pk_pets primary key (pet_id);
-
-alter table appointments
-alter column appointment_id set not null;
-alter table appointments
-add constraint pk_appointments primary key (appointment_id );
-----
+=== Store the password
 
+Ref:
 
-=== Store the password
+* https://docs.decodable.co/declarative/apply.html
+* https://docs.decodable.co/administer/manage-secrets.html
 
 [source,bash]
 ----
@@ -91,6 +73,8 @@ result: updated
 
 === Generate resource definitions
 
+Ref: https://docs.decodable.co/declarative/scan.html
+
 [source,bash]
 ----
 decodable connection scan \
@@ -107,29 +91,6 @@ decodable connection scan \
           > omd-pg.yaml
 ----
 
-=== Edit resource definitions
-
-
-- set to active
-
-[source,yaml]
-----
-spec_version: v2
-spec:
-    execution:
-      active: true
-----
-
-- add tags
-
-[source,yaml]
-----
-metadata:
-    tags:
-      project: "oh-my-dawg"
-      author: "rmoff"
-----
-
 === Apply resource definitions
 
 [source,bash]
@@ -161,3 +122,49 @@ id: 3cc8e060
 result: created
 ----
 
+=== Activate the Postgres connection
+
+[source,bash]
+----
+decodable query --name oh-my-dawg-pg -X activate --stabilize
+----
+
+== Sink connection: Snowflake
+
+=== Provision Snowflake resources
+
+Ref:
+
+* https://docs.decodable.co/connect/sink/snowflake.html
+* https://docs.snowflake.com/en/user-guide/key-pair-auth#verify-the-user-s-public-key-fingerprint
+
+=== Generate & create the Snowflake sink
+
+_You can pipe from one command to another to streamline the process._
+
+[source,bash]
+----
+decodable connection scan \
+          --name oh-my-dawg-snowflake \
+          --connector snowflake \
+          --type sink \
+          --prop snowflake.database=omd          \
+          --prop snowflake.schema=omd          \
+          --prop snowflake.user=decodable          \
+          --prop snowflake.private-key=$(decodable query --name omd-snowflake --keep-ids | yq '.metadata.id')          \
+          --prop snowflake.role=load_data          \
+          --prop snowflake.account-name=<org>-<account>          \
+          --prop snowflake.warehouse=stg          \
+          --prop snowflake.merge-interval="1 minute" \
+          --include-pattern stream-name='^omd-' \
+          | decodable apply -
+----
+
+NOTE: If you encounter errors then remove the pipe and `decodable apply` and inspect the YAML generated by `decodable connection scan` first.
+
+=== Activate the Snowflake connection
+
+[source,bash]
+----
+decodable query --name oh-my-dawg-snowflake -X activate --stabilize
+----

postgres-to-snowflake-with-cdc/decodable/omd-pg.yaml renamed to postgres-to-snowflake-with-cdc/decodable/sample_connection_resource_definitions/omd-pg.yaml

@@ -3,20 +3,17 @@ kind: connection
 metadata:
     name: oh-my-dawg-pg
     description: ""
-    tags:
-      project: "oh-my-dawg"
-      author: "rmoff"
 spec_version: v2
 spec:
     execution:
       active: true
     connector: postgres-cdc
     properties:
         database-name: postgres
-        hostname: 2.tcp.eu.ngrok.io
-        password: omd-pg
-        port: "19366"
+        hostname: <hostname>
+        port: <port>
         username: postgres
+        password: omd-pg
     stream_mappings:
         - stream_name: omd-appointments
           external_resource_specifier:
@@ -38,9 +35,6 @@ spec:
 kind: stream
 metadata:
     name: omd-appointments
-    tags:
-      project: "oh-my-dawg"
-      author: "rmoff"
 spec_version: v1
 spec:
     schema_v2:
@@ -79,9 +73,6 @@ spec:
 kind: stream
 metadata:
     name: omd-customers
-    tags:
-      project: "oh-my-dawg"
-      author: "rmoff"
 spec_version: v1
 spec:
     schema_v2:
@@ -111,9 +102,6 @@ spec:
 kind: stream
 metadata:
     name: omd-pets
-    tags:
-      project: "oh-my-dawg"
-      author: "rmoff"
 spec_version: v1
 spec:
     schema_v2:

postgres-to-snowflake-with-cdc/decodable/sample_connection_resource_definitions/omd-sf.yaml

@@ -0,0 +1,101 @@
+---
+kind: connection
+metadata:
+    name: oh-my-dawg-snowflake
+    description: ""
+spec_version: v2
+spec:
+    execution:
+      active: true
+    connector: snowflake
+    properties:
+      snowflake.account-name: <org_name>-<account_name>
+      snowflake.database: omd
+      snowflake.private-key: omd-snowflake
+      snowflake.role: load_data
+      snowflake.schema: omd
+      snowflake.user: decodable
+      snowflake.warehouse: stg
+      snowflake.merge-interval: 1 minute
+    stream_mappings:
+        - stream_name: omd-appointments
+          external_resource_specifier:
+            snowflake.table: omd-appointments
+        - stream_name: omd-pets
+          external_resource_specifier:
+            snowflake.table: omd-pets
+        - stream_name: omd-customers
+          external_resource_specifier:
+            snowflake.table: omd-customers
+    type: sink
+---
+kind: external-resource
+spec_version: v1
+spec:
+    connection: oh-my-dawg-snowflake
+    external_resource_specifier:
+        snowflake.table: omd-appointments
+    schema:
+        - name: appointment_id
+          type: VARCHAR NOT NULL
+        - name: duration_minutes
+          type: BIGINT
+        - name: appointment_time
+          type: VARCHAR
+        - name: status
+          type: VARCHAR
+        - name: pet_id
+          type: VARCHAR
+        - name: customer_id
+          type: VARCHAR
+        - name: service_type
+          type: VARCHAR
+        - name: price
+          type: DOUBLE
+        - name: last_updated
+          type: VARCHAR
+    configuration:
+        primary_key:
+            - appointment_id
+---
+kind: external-resource
+spec_version: v1
+spec:
+    connection: oh-my-dawg-snowflake
+    external_resource_specifier:
+        snowflake.table: omd-customers
+    schema:
+        - name: customer_id
+          type: VARCHAR NOT NULL
+        - name: first_name
+          type: VARCHAR
+        - name: last_name
+          type: VARCHAR
+        - name: phone
+          type: VARCHAR
+        - name: email
+          type: VARCHAR
+        - name: registered_date
+          type: VARCHAR
+    configuration:
+        primary_key:
+            - customer_id
+---
+kind: external-resource
+spec_version: v1
+spec:
+    connection: oh-my-dawg-snowflake
+    external_resource_specifier:
+        snowflake.table: omd-pets
+    schema:
+        - name: pet_id
+          type: VARCHAR NOT NULL
+        - name: customer_id
+          type: VARCHAR
+        - name: pet_name
+          type: VARCHAR
+        - name: pet_type
+          type: VARCHAR
+    configuration:
+        primary_key:
+            - pet_id

postgres-to-snowflake-with-cdc/decodable/sf-secret.yaml

@@ -0,0 +1,34 @@
+---
+kind: secret
+metadata:
+  name: omd-snowflake
+spec_version: v1
+spec:
+#  Don't work, this is a throwaway private key 😅
+  value_literal: |
+    MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQCRxsQUE8CdKQXu
+    Xz0Hcwjej5JYWAvgS68WsE/P1pwRjhE8bba5w3LmD1ijMlBShh3RvcM2gT9Cj3Lx
+    1X8D9pcdcOlwdOT3alEzXZjQlV7C4HC18xkEiEgNMjvWvP312af+Syq09u1UL4Hy
+    zl2TSI2XBTxttRmN6lY07Uk9ZHlGDjcL7+1ALg0Cq14PKFPNMnibKJ5cOWrtwZad
+    +HF6xTjyZRBDUpH72irHWIvwouLIM0ANhP6qTTaL5uxiSBxNyVcVKsgmaO2w1D+U
+    njHGL9lR55HPm1OoG6HOgI4KYDZ5xnVCQ0KCtiBHrrgtFOK00Gh31tBITxvlS7qs
+    kOGw1hz/AgMBAAECggEAAgJAr0KK0j+ZK6AK81oHXCj5uP27NpCK1AYcjqS9vgb8
+    pgRtbHrmgu56mxxO4gw9R44hxtPNUnZilmi9DHA7kQ+X9lF0dEcpR+Sfl5EI+MHm
+    zul2LBgd3tyupjaXGbbk0aRchswRfvb9B4kGquICXXt+sGGHl1G/qJDahDePU9x9
+    KE2WiMvXqbsdcD3U2RgFB8rSVQwx7vxLFH8islezyM3U7Q/G8LhWXtSEDoBR7vWD
+    uRpQ5+wqgBx8VdxJNt6dBYq5Umjag4FGnNTAEIJLqSfSHalofGMA/WjLjcb/YXth
+    o/Eyc+9U4+9rw90diM+tcBl3rr3UAHcLzC5WNoeVtQKBgQDIysJ3tcXgCZSs/TfU
+    aE1bgirtJxgRnvYBXKO1yQij20IsL5UM7YOCFxgzHmmjNuemKDeoCmKcU+tVFguE
+    ga4q9hhHTX8ZyqecgXUVwdMze67Sbv9T1eoG0smnvA+AjlElP5fKVUy9pBJI5+/5
+    +gKYilSfHxAj6N2hTnfp/OO2OwKBgQC525kcFSSNPw4Gt9VfaWlJtk0Oa6lrOn3H
+    4q00mYAcao5uIMAny7rW2KZB6YDB6X85y41XzecjOai/mAetNjNMR4KgDDMgP7JQ
+    yDjfHH8RxFcO86pxUriZeiU8X+dxRQQ7FXP/A5rCHtwZ+pj9YxE0/ZunDYxQmaC8
+    sbX7dEfUDQKBgG9aiviKlT9G8O3yzBh+84+xI487pAx5pKJituOkpqcAfLU2eime
+    OtVVa3VGA32hgFxUZ3FIuSFLJPKd9Cs7I9Ttf89jOf6atdOEs+MqB6/AgtZu+iiL
+    NGsuUOk10T8RLg1DNDHglluBdyZ5gkuWjAP+iylnt7LCfM7tTnE0bzBrAoGBALSL
+    zsCpCUjs6AM+sdht3gntPg20KHAx8d4rJXbjZsA0AwiYaBJAps/uxhNhceLtoNnU
+    Ewooy1A8wuDcHxj0fgCrtwki0MeTGPXAiv6x//6SbL/plLlhUlJFhcaQo5Q1J1b+
+    ECC6r6vDrqzN87CyfBSuCHbPgm8Jzkt/lvkejGhBAoGBAIAine7tBZ5G4bJxir3k
+    jA3tyO0TgdXUJz0KIJLiXeNf8K7xn+8CDu6UZ/WuDAMxMkwZdgZydufSWjhBGOXc
+    icGEqep9T0p/DPD6kz+HTMd1ad+Z+KtHA8G29NPwmn93TDAnVgQRIFh2LOHmwN74
+    TK5RPdOw9kN1/BXQpWQOdwzn

postgres-to-snowflake-with-cdc/docker-compose.yml

@@ -34,7 +34,7 @@ services:
     volumes:
       - ./shadowtraffic:/data
     command: --config /data/config.json
-#    --with-studio --sample 20 --watch
+#      --with-studio --sample 20 --watch
     ports:
 #      - 8080:8080
       - 9400:9400